fix(agents-runtime): avoid stack overflow on large live batches

.changeset/large-live-batches.md

@@ -0,0 +1,6 @@
+---
+"@electric-ax/agents-runtime": patch
+"@electric-ax/agents-server": patch
+---
+
+Avoid stack overflows when processing large live StreamDB batches and reading large JSON streams during server rehydration.

packages/agents-runtime/src/process-wake.ts

@@ -84,6 +84,12 @@ interface RawClaimCallbackResponse extends Omit<ClaimCallbackResponse, `ok`> {
 
 const DEFAULT_IDLE_TIMEOUT = 20_000
 const DEFAULT_HEARTBEAT_INTERVAL = 10_000
+
+function appendAll<T>(target: Array<T>, items: Array<T>): void {
+  for (const item of items) {
+    target.push(item)
+  }
+}
 type EntityStreamOptions = NonNullable<
   Parameters<typeof createEntityStreamDB>[3]
 >
@@ -1075,7 +1081,7 @@ export async function processWake(
       // ctx.events so no child result is silently acked without being visible.
       pendingLiveBatches.shift()
       batches.push(batch)
-      deltaEvents.push(...changeEvents)
+      appendAll(deltaEvents, changeEvents)
 
       if (freshKind === `inbox`) {
         selectedKind = `inbox`
@@ -1130,7 +1136,7 @@ export async function processWake(
     if (!preloaded) {
       const changeEvents = toChangeEvents(batch)
       if (changeEvents.length > 0) {
-        catchUpEvents.push(...changeEvents)
+        appendAll(catchUpEvents, changeEvents)
       }
       lastCatchUpOffset = batch.offset
       return
@@ -1145,7 +1151,7 @@ export async function processWake(
 
     handleLatestSignalEvents(changeEvents)
 
-    catchUpEvents.push(...changeEvents)
+    appendAll(catchUpEvents, changeEvents)
 
     if (
       resolveCurrentWakeReady !== null &&

packages/agents-server/src/stream-client.ts

@@ -417,7 +417,11 @@ export class StreamClient {
         offset: fromOffset ?? `-1`,
         live: false,
       })
-      return await response.json<T>()
+      const items: Array<T> = []
+      for await (const item of response.jsonStream()) {
+        items.push(item)
+      }
+      return items
     })
   }
 

packages/agents-server/test/stream-client.test.ts

@@ -7,6 +7,7 @@ const {
   flushMock,
   detachMock,
   headMock,
+  streamMock,
   MockFetchError,
   MockDurableStreamError,
 } = vi.hoisted(() => {
@@ -29,6 +30,7 @@ const {
     flushMock: vi.fn().mockResolvedValue(undefined),
     detachMock: vi.fn().mockResolvedValue(undefined),
     headMock: vi.fn(),
+    streamMock: vi.fn(),
     MockFetchError: HoistedFetchError,
     MockDurableStreamError: HoistedDurableStreamError,
   }
@@ -38,6 +40,8 @@ vi.mock(`@durable-streams/client`, () => ({
   DurableStream: class {
     constructor(_opts: { url: string; contentType?: string }) {}
 
+    stream = streamMock
+
     static create = vi.fn()
     static delete = vi.fn()
     static head = headMock
@@ -58,6 +62,31 @@ describe(`StreamClient`, () => {
     flushMock.mockClear()
     detachMock.mockClear()
     headMock.mockReset()
+    streamMock.mockReset()
+  })
+
+  it(`readJson reads large batches without using StreamResponse.json`, async () => {
+    const largeBatch = Array.from({ length: 70_000 }, (_, i) => ({ i }))
+    const jsonMock = vi.fn(() => {
+      throw new RangeError(`Maximum call stack size exceeded`)
+    })
+    streamMock.mockResolvedValueOnce({
+      json: jsonMock,
+      async *jsonStream() {
+        for (const item of largeBatch) {
+          yield item
+        }
+      },
+    })
+
+    const client = new StreamClient(`http://127.0.0.1:4545`)
+
+    await expect(client.readJson(`/large-json`)).resolves.toEqual(largeBatch)
+    expect(jsonMock).not.toHaveBeenCalled()
+    expect(streamMock).toHaveBeenCalledWith({
+      offset: `-1`,
+      live: false,
+    })
   })
 
   it(`appendIdempotent uses IdempotentProducer append/flush/detach`, async () => {