feat(server): Gracefully handle oversized batch items instead of aborting the stream (#3137)

Gracefully handle oversized batch items instead of aborting the stream. When an NDJSON batch item exceeds the maximum size, the parser now emits an error marker instead of throwing, allowing the batch to seal normally. The oversized item becomes a pre-failed run with `PAYLOAD_TOO_LARGE` error code, while other items in the batch process successfully. This prevents `batchTriggerAndWait` from seeing connection errors and retrying with exponential backoff. Also fixes the NDJSON parser not consuming the remainder of an oversized line split across multiple chunks, which caused "Invalid JSON" errors on subsequent lines.
2026-02-27 10:11:42 +00:00
parent cff45664fc
commit 8003923598
14 changed files with 541 additions and 86 deletions
@@ -0,0 +1,5 @@
+---
+"@trigger.dev/sdk": patch
+---
+
+Add PAYLOAD_TOO_LARGE error to handle graceful recovery of sending batch trigger items with payloads that exceed the maximum payload size
@@ -0,0 +1,10 @@
+---
+area: webapp
+type: fix
+---
+
+Gracefully handle oversized batch items instead of aborting the stream.
+
+When an NDJSON batch item exceeds the maximum size, the parser now emits an error marker instead of throwing, allowing the batch to seal normally. The oversized item becomes a pre-failed run with `PAYLOAD_TOO_LARGE` error code, while other items in the batch process successfully. This prevents `batchTriggerAndWait` from seeing connection errors and retrying with exponential backoff.
+
+Also fixes the NDJSON parser not consuming the remainder of an oversized line split across multiple chunks, which caused "Invalid JSON" errors on subsequent lines.
@@ -509,8 +509,7 @@ export class SpanPresenter extends BasePresenter {
        taskIdentifier: true,
        spanId: true,
        createdAt: true,
-        number: true,
-        taskVersion: true,
+        status: true,
      },
      where: {
        parentSpanId: spanId,
@@ -99,11 +99,8 @@ export async function action({ request, params }: ActionFunctionArgs) {
    if (error instanceof ServiceValidationError) {
      return json({ error: error.message }, { status: 422 });
    } else if (error instanceof Error) {
-      // Check for stream parsing errors
-      if (
-        error.message.includes("Invalid JSON") ||
-        error.message.includes("exceeds maximum size")
-      ) {
+      // Check for stream parsing errors (e.g. invalid JSON)
+      if (error.message.includes("Invalid JSON")) {
        return json({ error: error.message }, { status: 400 });
      }

@@ -58,6 +58,7 @@ import { RunTimeline, RunTimelineEvent, SpanTimeline } from "~/components/run/Ru
 import { PacketDisplay } from "~/components/runs/v3/PacketDisplay";
 import { RunIcon } from "~/components/runs/v3/RunIcon";
 import { RunTag } from "~/components/runs/v3/RunTag";
+import { TruncatedCopyableValue } from "~/components/primitives/TruncatedCopyableValue";
 import { SpanEvents } from "~/components/runs/v3/SpanEvents";
 import { SpanTitle } from "~/components/runs/v3/SpanTitle";
 import { TaskRunAttemptStatusCombo } from "~/components/runs/v3/TaskRunAttemptStatus";
@@ -133,7 +134,8 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => {
              name: error.name,
              message: error.message,
              stack: error.stack,
-              cause: error.cause instanceof Error
+              cause:
+                error.cause instanceof Error
                  ? { name: error.cause.name, message: error.cause.message }
                  : error.cause,
            }
@@ -1003,7 +1005,7 @@ function RunBody({
          )}
        </div>
      </div>
-      <div className="flex items-center flex-wrap py-2 justify-between gap-2 border-t border-grid-dimmed px-2">
+      <div className="flex flex-wrap items-center justify-between gap-2 border-t border-grid-dimmed px-2 py-2">
        <div className="flex items-center gap-4">
          {run.friendlyId !== runParam && (
            <LinkButton
@@ -1047,9 +1049,11 @@ function RunBody({
                  </PopoverTrigger>
                  <PopoverContent className="min-w-[140px] p-1" align="end">
                    <PopoverMenuItem
-                      to={`${v3LogsPath(organization, project, environment)}?runId=${runParam}&from=${
-                        new Date(run.createdAt).getTime() - 60000
-                      }`}
+                      to={`${v3LogsPath(
+                        organization,
+                        project,
+                        environment
+                      )}?runId=${runParam}&from=${new Date(run.createdAt).getTime() - 60000}`}
                      title="View logs"
                      icon={ArrowRightIcon}
                      leadingIconClassName="text-blue-500"
@@ -1200,50 +1204,6 @@ function SpanEntity({ span }: { span: Span }) {
            <Property.Label>Message</Property.Label>
            <Property.Value className="whitespace-pre-wrap">{span.message}</Property.Value>
          </Property.Item>
-          {span.triggeredRuns.length > 0 && (
-            <Property.Item>
-              <div className="flex flex-col gap-1.5">
-                <Header3>Triggered runs</Header3>
-                <Table containerClassName="max-h-[12.5rem]">
-                  <TableHeader className="bg-background-bright">
-                    <TableRow>
-                      <TableHeaderCell>Run #</TableHeaderCell>
-                      <TableHeaderCell>Task</TableHeaderCell>
-                      <TableHeaderCell>Version</TableHeaderCell>
-                      <TableHeaderCell>Created at</TableHeaderCell>
-                    </TableRow>
-                  </TableHeader>
-                  <TableBody>
-                    {span.triggeredRuns.map((run) => {
-                      const path = v3RunSpanPath(
-                        organization,
-                        project,
-                        environment,
-                        { friendlyId: run.friendlyId },
-                        { spanId: run.spanId }
-                      );
-                      return (
-                        <TableRow key={run.friendlyId}>
-                          <TableCell to={path} actionClassName="py-1.5" rowHoverStyle="bright">
-                            {run.number}
-                          </TableCell>
-                          <TableCell to={path} actionClassName="py-1.5" rowHoverStyle="bright">
-                            {run.taskIdentifier}
-                          </TableCell>
-                          <TableCell to={path} actionClassName="py-1.5" rowHoverStyle="bright">
-                            {run.taskVersion ?? "–"}
-                          </TableCell>
-                          <TableCell to={path} actionClassName="py-1.5" rowHoverStyle="bright">
-                            <DateTime date={run.createdAt} />
-                          </TableCell>
-                        </TableRow>
-                      );
-                    })}
-                  </TableBody>
-                </Table>
-              </div>
-            </Property.Item>
-          )}
        </Property.Table>
        {span.events.length > 0 && <SpanEvents spanEvents={span.events} />}
        {span.properties !== undefined ? (
@@ -1268,6 +1228,48 @@ function SpanEntity({ span }: { span: Span }) {
            showOpenInModal
          />
        ) : null}
+        {span.triggeredRuns.length > 0 && (
+          <div className="flex flex-col gap-1.5">
+            <Header3>Runs</Header3>
+            <Table containerClassName="max-h-[12.5rem]">
+              <TableHeader className="bg-background-bright">
+                <TableRow>
+                  <TableHeaderCell>ID</TableHeaderCell>
+                  <TableHeaderCell>Task</TableHeaderCell>
+                  <TableHeaderCell>Status</TableHeaderCell>
+                  <TableHeaderCell>Created</TableHeaderCell>
+                </TableRow>
+              </TableHeader>
+              <TableBody>
+                {span.triggeredRuns.map((run) => {
+                  const path = v3RunSpanPath(
+                    organization,
+                    project,
+                    environment,
+                    { friendlyId: run.friendlyId },
+                    { spanId: run.spanId }
+                  );
+                  return (
+                    <TableRow key={run.friendlyId}>
+                      <TableCell to={path} actionClassName="py-1.5" rowHoverStyle="bright">
+                        <TruncatedCopyableValue value={run.friendlyId} />
+                      </TableCell>
+                      <TableCell to={path} actionClassName="py-1.5" rowHoverStyle="bright">
+                        {run.taskIdentifier}
+                      </TableCell>
+                      <TableCell to={path} actionClassName="py-1.5" rowHoverStyle="bright">
+                        <TaskRunStatusCombo status={run.status} />
+                      </TableCell>
+                      <TableCell to={path} actionClassName="py-1.5" rowHoverStyle="bright">
+                        <DateTimeAccurate date={run.createdAt} hour12={false} hideDate={true} />
+                      </TableCell>
+                    </TableRow>
+                  );
+                })}
+              </TableBody>
+            </Table>
+          </div>
+        )}
      </div>
    );
  }
@@ -14,6 +14,14 @@ export type StreamBatchItemsServiceOptions = {
  maxItemBytes: number;
 };

+export type OversizedItemMarker = {
+  __batchItemError: "OVERSIZED";
+  index: number;
+  task: string;
+  actualSize: number;
+  maxSize: number;
+};
+
 export type StreamBatchItemsServiceConstructorOptions = {
  prisma?: PrismaClientOrTransaction;
  engine?: RunEngine;
@@ -110,6 +118,41 @@ export class StreamBatchItemsService extends WithRunEngine {

        // Process items from the stream
        for await (const rawItem of itemsIterator) {
+          // Check for oversized item markers from the NDJSON parser
+          if (rawItem && typeof rawItem === "object" && "__batchItemError" in rawItem) {
+            const marker = rawItem as OversizedItemMarker;
+            const itemIndex = marker.index >= 0 ? marker.index : lastIndex + 1;
+
+            const errorMessage = `Batch item payload is too large (${(marker.actualSize / 1024).toFixed(1)} KB). Maximum allowed size is ${(marker.maxSize / 1024).toFixed(1)} KB. Reduce the payload size or offload large data to external storage.`;
+
+            // Enqueue with __error metadata - processItemCallback will detect this
+            // and use TriggerFailedTaskService to create a pre-failed run
+            const batchItem: BatchItem = {
+              task: marker.task,
+              payload: "{}",
+              payloadType: "application/json",
+              options: {
+                __error: errorMessage,
+                __errorCode: "PAYLOAD_TOO_LARGE",
+              },
+            };
+
+            const result = await this._engine.enqueueBatchItem(
+              batchId,
+              environment.id,
+              itemIndex,
+              batchItem
+            );
+
+            if (result.enqueued) {
+              itemsAccepted++;
+            } else {
+              itemsDeduplicated++;
+            }
+            lastIndex = itemIndex;
+            continue;
+          }
+
          // Parse and validate the item
          const parseResult = BatchItemNDJSONSchema.safeParse(rawItem);
          if (!parseResult.success) {
@@ -168,6 +211,34 @@ export class StreamBatchItemsService extends WithRunEngine {

        // Validate we received the expected number of items
        if (enqueuedCount !== batch.runCount) {
+          // The batch queue consumers may have already processed all items and
+          // cleaned up the Redis keys before we got here (especially likely when
+          // items include pre-failed runs that complete instantly). Check if the
+          // batch was already sealed/completed in Postgres.
+          const currentBatch = await this._prisma.batchTaskRun.findUnique({
+            where: { id: batchId },
+            select: { sealed: true, status: true },
+          });
+
+          if (currentBatch?.sealed) {
+            logger.info("Batch already sealed before count check (fast completion)", {
+              batchId: batchFriendlyId,
+              itemsAccepted,
+              itemsDeduplicated,
+              enqueuedCount,
+              expectedCount: batch.runCount,
+              batchStatus: currentBatch.status,
+            });
+
+            return {
+              id: batchFriendlyId,
+              itemsAccepted,
+              itemsDeduplicated,
+              sealed: true,
+              runCount: batch.runCount,
+            };
+          }
+
          logger.warn("Batch item count mismatch", {
            batchId: batchFriendlyId,
            expected: batch.runCount,
@@ -281,6 +352,121 @@ export class StreamBatchItemsService extends WithRunEngine {
  }
 }

+/**
+ * Extract `index` and `task` from raw JSON bytes without decoding the full line.
+ * Scans at most 512 bytes, tracking JSON nesting depth to only match top-level keys.
+ */
+export function extractIndexAndTask(bytes: Uint8Array): { index: number; task: string } {
+  let index = -1;
+  let task = "unknown";
+  let depth = 0;
+  let foundIndex = false;
+  let foundTask = false;
+  const limit = Math.min(bytes.byteLength, 512);
+
+  const QUOTE = 0x22; // "
+  const COLON = 0x3a; // :
+  const LBRACE = 0x7b; // {
+  const RBRACE = 0x7d; // }
+  const LBRACKET = 0x5b; // [
+  const RBRACKET = 0x5d; // ]
+  const BACKSLASH = 0x5c; // \
+
+  // Byte patterns for "index" and "task" (without quotes)
+  const INDEX_BYTES = [0x69, 0x6e, 0x64, 0x65, 0x78]; // index
+  const TASK_BYTES = [0x74, 0x61, 0x73, 0x6b]; // task
+
+  let i = 0;
+  while (i < limit && !(foundIndex && foundTask)) {
+    const b = bytes[i];
+
+    if (b === LBRACE || b === LBRACKET) {
+      depth++;
+      i++;
+      continue;
+    }
+    if (b === RBRACE || b === RBRACKET) {
+      depth--;
+      i++;
+      continue;
+    }
+
+    // Only match keys at depth 1 (top-level object)
+    if (b === QUOTE && depth === 1) {
+      // Read the key inside quotes
+      const keyStart = i + 1;
+      let keyEnd = keyStart;
+      while (keyEnd < limit && bytes[keyEnd] !== QUOTE) {
+        if (bytes[keyEnd] === BACKSLASH) keyEnd++; // skip escaped char
+        keyEnd++;
+      }
+
+      const keyLen = keyEnd - keyStart;
+
+      // Check if this key matches "index" or "task"
+      const isIndex =
+        !foundIndex &&
+        keyLen === INDEX_BYTES.length &&
+        INDEX_BYTES.every((b, j) => bytes[keyStart + j] === b);
+      const isTask =
+        !foundTask &&
+        keyLen === TASK_BYTES.length &&
+        TASK_BYTES.every((b, j) => bytes[keyStart + j] === b);
+
+      if (isIndex || isTask) {
+        // Skip past closing quote and find colon
+        let pos = keyEnd + 1;
+        while (pos < limit && bytes[pos] !== COLON) pos++;
+        pos++; // skip colon
+        // Skip whitespace
+        while (pos < limit && (bytes[pos] === 0x20 || bytes[pos] === 0x09)) pos++;
+
+        if (isIndex) {
+          // Parse digits
+          let num = 0;
+          let hasDigit = false;
+          while (pos < limit && bytes[pos] >= 0x30 && bytes[pos] <= 0x39) {
+            num = num * 10 + (bytes[pos] - 0x30);
+            hasDigit = true;
+            pos++;
+          }
+          if (hasDigit) {
+            index = num;
+            foundIndex = true;
+          }
+        } else {
+          // Parse quoted string value
+          if (pos < limit && bytes[pos] === QUOTE) {
+            const valStart = pos + 1;
+            let valEnd = valStart;
+            while (valEnd < limit && bytes[valEnd] !== QUOTE) {
+              if (bytes[valEnd] === BACKSLASH) valEnd++;
+              valEnd++;
+            }
+            // Decode just this slice
+            try {
+              task = new TextDecoder("utf-8", { fatal: true }).decode(
+                bytes.slice(valStart, valEnd)
+              );
+              foundTask = true;
+            } catch {
+              // Leave as "unknown"
+            }
+          }
+        }
+      }
+
+      // Skip past the key's closing quote
+      i = keyEnd + 1;
+      continue;
+    }
+
+    i++;
+  }
+
+  return { index, task };
+}
+
 /**
 * Create an NDJSON parser transform stream.
 *
@@ -305,6 +491,9 @@ export function createNdjsonParserStream(
  let chunks: Uint8Array[] = [];
  let totalBytes = 0;
  let lineNumber = 0;
+  // When an oversized incomplete line is detected (Case 2), we must discard
+  // all remaining bytes of that line until the next newline delimiter.
+  let skipUntilNewline = false;

  const NEWLINE_BYTE = 0x0a; // '\n'

@@ -398,6 +587,24 @@ export function createNdjsonParserStream(

  return new TransformStream<Uint8Array, unknown>({
    transform(chunk, controller) {
+      // If we're skipping the remainder of an oversized line, scan for the
+      // next newline in this chunk and discard everything before it.
+      if (skipUntilNewline) {
+        const nlPos = chunk.indexOf(NEWLINE_BYTE);
+        if (nlPos === -1) {
+          // Entire chunk is still part of the oversized line — discard it
+          return;
+        }
+        // Found the newline — keep everything after it
+        skipUntilNewline = false;
+        const remaining = chunk.slice(nlPos + 1);
+        if (remaining.byteLength === 0) {
+          return;
+        }
+        // Replace chunk with the remainder and fall through to normal processing
+        chunk = remaining;
+      }
+
      // Append chunk to buffer
      chunks.push(chunk);
      totalBytes += chunk.byteLength;
@@ -407,11 +614,19 @@ export function createNdjsonParserStream(
      while ((newlineIndex = findNewlineIndex()) !== -1) {
        // Check size limit BEFORE extracting/decoding (bytes up to newline)
        if (newlineIndex > maxItemBytes) {
-          throw new Error(
-            `Item at line ${
-              lineNumber + 1
-            } exceeds maximum size of ${maxItemBytes} bytes (actual: ${newlineIndex})`
-          );
+          // Case 1: Complete line exceeds limit - emit marker instead of throwing
+          const lineBytes = extractLine(newlineIndex);
+          const extracted = extractIndexAndTask(lineBytes);
+          const marker: OversizedItemMarker = {
+            __batchItemError: "OVERSIZED",
+            index: extracted.index,
+            task: extracted.task,
+            actualSize: newlineIndex,
+            maxSize: maxItemBytes,
+          };
+          controller.enqueue(marker);
+          lineNumber++;
+          continue;
        }

        const lineBytes = extractLine(newlineIndex);
@@ -421,11 +636,23 @@ export function createNdjsonParserStream(
      // Check if the remaining buffer (incomplete line) exceeds the limit
      // This prevents OOM from a single huge line without newlines
      if (totalBytes > maxItemBytes) {
-        throw new Error(
-          `Item at line ${
-            lineNumber + 1
-          } exceeds maximum size of ${maxItemBytes} bytes (buffered: ${totalBytes}, no newline found)`
-        );
+        // Case 2: Incomplete line exceeds limit - emit marker instead of throwing
+        const extracted = extractIndexAndTask(concatenateChunks());
+        const marker: OversizedItemMarker = {
+          __batchItemError: "OVERSIZED",
+          index: extracted.index,
+          task: extracted.task,
+          actualSize: totalBytes,
+          maxSize: maxItemBytes,
+        };
+        controller.enqueue(marker);
+        lineNumber++;
+        // Clear buffer and skip remaining bytes of this oversized line
+        // until the next newline delimiter is found in a subsequent chunk
+        chunks = [];
+        totalBytes = 0;
+        skipUntilNewline = true;
+        return;
      }
    },

@@ -441,11 +668,17 @@ export function createNdjsonParserStream(

      // Check size limit before processing final line
      if (totalBytes > maxItemBytes) {
-        throw new Error(
-          `Item at line ${
-            lineNumber + 1
-          } exceeds maximum size of ${maxItemBytes} bytes (actual: ${totalBytes})`
-        );
+        // Case 3: Flush with oversized remaining - emit marker instead of throwing
+        const extracted = extractIndexAndTask(concatenateChunks());
+        const marker: OversizedItemMarker = {
+          __batchItemError: "OVERSIZED",
+          index: extracted.index,
+          task: extracted.task,
+          actualSize: totalBytes,
+          maxSize: maxItemBytes,
+        };
+        controller.enqueue(marker);
+        return;
      }

      const finalBytes = concatenateChunks();
@@ -669,6 +669,46 @@ export function setupBatchQueueCallbacks() {
          engine,
        });

+        // Check for pre-marked error items (e.g. oversized payloads)
+        const itemError = item.options?.__error as string | undefined;
+        if (itemError) {
+          const errorCode = (item.options?.__errorCode as string) ?? "ITEM_ERROR";
+
+          let environment: AuthenticatedEnvironment | undefined;
+          try {
+            environment = (await findEnvironmentById(meta.environmentId)) ?? undefined;
+          } catch {
+            // Best-effort environment lookup
+          }
+
+          if (environment) {
+            const failedRunId = await triggerFailedTaskService.call({
+              taskId: item.task,
+              environment,
+              payload: item.payload ?? "{}",
+              payloadType: item.payloadType as string,
+              errorMessage: itemError,
+              errorCode: errorCode as TaskRunErrorCodes,
+              parentRunId: meta.parentRunId,
+              resumeParentOnCompletion: meta.resumeParentOnCompletion,
+              batch: { id: batchId, index: itemIndex },
+              traceContext: meta.traceContext as Record<string, unknown> | undefined,
+              spanParentAsLink: meta.spanParentAsLink,
+            });
+
+            if (failedRunId) {
+              span.setAttribute("batch.result.pre_failed", true);
+              span.setAttribute("batch.result.run_id", failedRunId);
+              span.end();
+              return { success: true as const, runId: failedRunId };
+            }
+          }
+
+          // Fallback if TriggerFailedTaskService or environment lookup fails
+          span.end();
+          return { success: false as const, error: itemError, errorCode };
+        }
+
        let environment: AuthenticatedEnvironment | undefined;
        try {
          environment = (await findEnvironmentById(meta.environmentId)) ?? undefined;
@@ -24,6 +24,8 @@ import {
  StreamBatchItemsService,
  createNdjsonParserStream,
  streamToAsyncIterable,
+  extractIndexAndTask,
+  type OversizedItemMarker,
 } from "../../app/runEngine/services/streamBatchItems.server";
 import { ServiceValidationError } from "../../app/v3/services/baseService.server";

@@ -641,6 +643,25 @@ describe("createNdjsonParserStream", () => {
    expect(results).toEqual([{ id: 1 }, { id: 2 }]);
  });

+  it("should handle escaped newlines in JSON string values", async () => {
+    // JSON.stringify escapes newlines as \n (two chars: backslash + n),
+    // so they don't break NDJSON line boundaries. This is the normal case
+    // when the SDK serializes payloads containing newlines.
+    const item1 = JSON.stringify({ payload: "line1\nline2\nline3" });
+    const item2 = JSON.stringify({ payload: "no newlines" });
+    const ndjson = item1 + "\n" + item2 + "\n";
+    const encoder = new TextEncoder();
+    const stream = chunksToStream([encoder.encode(ndjson)]);
+
+    const parser = createNdjsonParserStream(1024);
+    const results = await collectStream(stream.pipeThrough(parser));
+
+    expect(results).toEqual([
+      { payload: "line1\nline2\nline3" },
+      { payload: "no newlines" },
+    ]);
+  });
+
  it("should skip empty lines", async () => {
    const ndjson = '{"a":1}\n\n{"b":2}\n   \n{"c":3}\n';
    const encoder = new TextEncoder();
@@ -705,33 +726,76 @@ describe("createNdjsonParserStream", () => {
    expect(results).toEqual([{ greeting: "こんにちは" }]);
  });

-  it("should reject lines exceeding maxItemBytes", async () => {
+  it("should emit OversizedItemMarker for lines exceeding maxItemBytes", async () => {
    const maxBytes = 50;
-    // Create a line that exceeds the limit
-    const largeJson = JSON.stringify({ data: "x".repeat(100) }) + "\n";
+    // Create a line that exceeds the limit with index and task fields
+    const largeJson = JSON.stringify({ index: 3, task: "my-task", data: "x".repeat(100) }) + "\n";
    const encoder = new TextEncoder();
    const stream = chunksToStream([encoder.encode(largeJson)]);

    const parser = createNdjsonParserStream(maxBytes);
+    const results = await collectStream(stream.pipeThrough(parser));

-    await expect(collectStream(stream.pipeThrough(parser))).rejects.toThrow(/exceeds maximum size/);
+    expect(results).toHaveLength(1);
+    const marker = results[0] as OversizedItemMarker;
+    expect(marker.__batchItemError).toBe("OVERSIZED");
+    expect(marker.index).toBe(3);
+    expect(marker.task).toBe("my-task");
+    expect(marker.maxSize).toBe(maxBytes);
+    expect(marker.actualSize).toBeGreaterThan(maxBytes);
  });

-  it("should reject unbounded accumulation without newlines", async () => {
+  it("should emit OversizedItemMarker for unbounded accumulation without newlines", async () => {
    const maxBytes = 50;
    // Send data without any newlines that exceeds the buffer limit
    const encoder = new TextEncoder();
    const chunks = [
-      encoder.encode('{"start":"'),
+      encoder.encode('{"index":7,"task":"big-task","start":"'),
      encoder.encode("x".repeat(60)), // This will push buffer over 50 bytes
    ];
    const stream = chunksToStream(chunks);

    const parser = createNdjsonParserStream(maxBytes);
+    const results = await collectStream(stream.pipeThrough(parser));

-    await expect(collectStream(stream.pipeThrough(parser))).rejects.toThrow(
-      /exceeds maximum size.*no newline found/
-    );
+    expect(results).toHaveLength(1);
+    const marker = results[0] as OversizedItemMarker;
+    expect(marker.__batchItemError).toBe("OVERSIZED");
+    expect(marker.index).toBe(7);
+    expect(marker.task).toBe("big-task");
+    expect(marker.maxSize).toBe(maxBytes);
+  });
+
+  it("should skip remaining bytes of oversized line arriving in subsequent chunks", async () => {
+    const maxBytes = 50;
+    const encoder = new TextEncoder();
+    // Simulate a normal item, then an oversized item split across many chunks,
+    // then another normal item after the newline.
+    // The oversized line is: {"index":1,"task":"t","data":"xxxx...120 x's...xxxx"}\n
+    const normalItem1 = '{"index":0,"task":"t","x":1}\n';
+    const oversizedStart = '{"index":1,"task":"t","data":"';
+    const oversizedMiddle = "x".repeat(120); // way over 50 bytes
+    const oversizedEnd = '"}\n';
+    const normalItem2 = '{"index":2,"task":"t","x":2}\n';
+
+    // Send as separate chunks to trigger Case 2 (no newline, buffer > limit)
+    const chunks = [
+      encoder.encode(normalItem1 + oversizedStart),
+      encoder.encode(oversizedMiddle.slice(0, 60)),
+      encoder.encode(oversizedMiddle.slice(60)),
+      encoder.encode(oversizedEnd + normalItem2),
+    ];
+    const stream = chunksToStream(chunks);
+
+    const parser = createNdjsonParserStream(maxBytes);
+    const results = await collectStream(stream.pipeThrough(parser));
+
+    // Should get: normal item 1, oversized marker, normal item 2
+    expect(results).toHaveLength(3);
+    expect(results[0]).toEqual({ index: 0, task: "t", x: 1 });
+    expect((results[1] as OversizedItemMarker).__batchItemError).toBe("OVERSIZED");
+    expect((results[1] as OversizedItemMarker).index).toBe(1);
+    expect(results[2]).toEqual({ index: 2, task: "t", x: 2 });
  });

  it("should check byte size before decoding to prevent OOM", async () => {
@@ -756,10 +820,12 @@ describe("createNdjsonParserStream", () => {
    const results1 = await collectStream(stream1.pipeThrough(parser1));
    expect(results1).toHaveLength(1);

-    // Large one should fail
+    // Large one should emit an OversizedItemMarker
    const stream2 = chunksToStream([largeBytes]);
    const parser2 = createNdjsonParserStream(maxBytes);
-    await expect(collectStream(stream2.pipeThrough(parser2))).rejects.toThrow(/exceeds maximum/);
+    const results2 = await collectStream(stream2.pipeThrough(parser2));
+    expect(results2).toHaveLength(1);
+    expect((results2[0] as OversizedItemMarker).__batchItemError).toBe("OVERSIZED");
  });

  it("should handle final line in flush without trailing newline", async () => {
@@ -837,6 +903,28 @@ describe("createNdjsonParserStream", () => {
    expect(results).toEqual([]);
  });

+  it("should pass normal items and emit markers for oversized items in the same stream", async () => {
+    const maxBytes = 50;
+    const encoder = new TextEncoder();
+    // Normal item, then oversized item, then another normal item
+    const normalItem1 = '{"index":0,"task":"t","x":1}\n';
+    const oversizedItem = JSON.stringify({ index: 1, task: "t", data: "x".repeat(100) }) + "\n";
+    const normalItem2 = '{"index":2,"task":"t","x":2}\n';
+    const stream = chunksToStream([encoder.encode(normalItem1 + oversizedItem + normalItem2)]);
+
+    const parser = createNdjsonParserStream(maxBytes);
+    const results = await collectStream(stream.pipeThrough(parser));
+
+    expect(results).toHaveLength(3);
+    // First: normal parsed object
+    expect(results[0]).toEqual({ index: 0, task: "t", x: 1 });
+    // Second: oversized marker
+    expect((results[1] as OversizedItemMarker).__batchItemError).toBe("OVERSIZED");
+    expect((results[1] as OversizedItemMarker).index).toBe(1);
+    // Third: normal parsed object
+    expect(results[2]).toEqual({ index: 2, task: "t", x: 2 });
+  });
+
  it("should handle stream with only whitespace", async () => {
    const encoder = new TextEncoder();
    const stream = chunksToStream([encoder.encode("   \n\n   \n")]);
@@ -847,3 +935,34 @@ describe("createNdjsonParserStream", () => {
    expect(results).toEqual([]);
  });
 });
+
+describe("extractIndexAndTask", () => {
+  const encoder = new TextEncoder();
+
+  it("should extract index and task from JSON bytes", () => {
+    const bytes = encoder.encode('{"index":42,"task":"my-task","data":"x"}');
+    const result = extractIndexAndTask(bytes);
+    expect(result.index).toBe(42);
+    expect(result.task).toBe("my-task");
+  });
+
+  it("should return defaults for empty or malformed bytes", () => {
+    const result = extractIndexAndTask(new Uint8Array(0));
+    expect(result.index).toBe(-1);
+    expect(result.task).toBe("unknown");
+  });
+
+  it("should handle keys in any order", () => {
+    const bytes = encoder.encode('{"task":"other-task","data":"y","index":99}');
+    const result = extractIndexAndTask(bytes);
+    expect(result.index).toBe(99);
+    expect(result.task).toBe("other-task");
+  });
+
+  it("should not match nested keys", () => {
+    const bytes = encoder.encode('{"nested":{"index":999,"task":"inner"},"index":5,"task":"outer"}');
+    const result = extractIndexAndTask(bytes);
+    expect(result.index).toBe(5);
+    expect(result.task).toBe("outer");
+  });
+});
@@ -848,7 +848,11 @@ export class BatchQueue {
            "BatchQueue.serializePayload",
            async (innerSpan) => {
              const str =
-                typeof item.payload === "string" ? item.payload : JSON.stringify(item.payload);
+                item.payload === undefined || item.payload === null
+                  ? "{}"
+                  : typeof item.payload === "string"
+                    ? item.payload
+                    : JSON.stringify(item.payload);
              innerSpan?.setAttribute("batch.payloadSize", str.length);
              return str;
            }
@@ -912,7 +916,11 @@ export class BatchQueue {
          "BatchQueue.serializePayload",
          async (innerSpan) => {
            const str =
-              typeof item.payload === "string" ? item.payload : JSON.stringify(item.payload);
+              item.payload === undefined || item.payload === null
+                ? "{}"
+                : typeof item.payload === "string"
+                  ? item.payload
+                  : JSON.stringify(item.payload);
            innerSpan?.setAttribute("batch.payloadSize", str.length);
            return str;
          }
@@ -61,6 +61,7 @@ export function runStatusFromError(
    case "TASK_PROCESS_SIGTERM":
    case "TASK_DID_CONCURRENT_WAIT":
    case "BATCH_ITEM_COULD_NOT_TRIGGER":
+    case "PAYLOAD_TOO_LARGE":
    case "UNSPECIFIED_ERROR":
      return "SYSTEM_FAILURE";
    default:
@@ -308,6 +308,7 @@ export function shouldRetryError(error: TaskRunError): boolean {
        case "TASK_HAS_N0_EXECUTION_SNAPSHOT":
        case "TASK_RUN_DEQUEUED_MAX_RETRIES":
        case "BATCH_ITEM_COULD_NOT_TRIGGER":
+        case "PAYLOAD_TOO_LARGE":
        case "UNSPECIFIED_ERROR":
          return false;

@@ -188,6 +188,7 @@ export const TaskRunInternalError = z.object({
    "TASK_DID_CONCURRENT_WAIT",
    "RECURSIVE_WAIT_DEADLOCK",
    "BATCH_ITEM_COULD_NOT_TRIGGER",
+    "PAYLOAD_TOO_LARGE",
    "UNSPECIFIED_ERROR",
  ]),
  message: z.string().optional(),
@@ -999,6 +999,45 @@ export const largePayloadTask = task({
  },
 });

+// ============================================================================
+// Oversized Payload Graceful Handling
+// ============================================================================
+
+/**
+ * Test: Batch with oversized item should complete gracefully
+ *
+ * Sends 2 items: one normal, one oversized (~3.2MB).
+ * The oversized item should result in a pre-failed run (ok: false)
+ * while the normal item processes successfully (ok: true).
+ */
+export const batchSealFailureOversizedPayload = task({
+  id: "batch-seal-failure-oversized",
+  maxDuration: 60,
+  retry: {
+    maxAttempts: 1,
+  },
+  run: async () => {
+    const results = await fixedLengthTask.batchTriggerAndWait([
+      { payload: { waitSeconds: 1, output: "normal" } },
+      { payload: { waitSeconds: 1, output: "x".repeat(3_200_000) } }, // ~3.2MB oversized
+    ]);
+
+    const normal = results.runs[0];
+    const oversized = results.runs[1];
+
+    logger.info("Batch results", {
+      normalOk: normal?.ok,
+      oversizedOk: oversized?.ok,
+    });
+
+    return {
+      normalOk: normal?.ok === true,
+      oversizedOk: oversized?.ok === false,
+      oversizedError: !oversized?.ok ? oversized?.error : undefined,
+    };
+  },
+});
+
 type Payload = {
  waitSeconds: number;
  error?: string;