📉 perf: Bound Early Event Buffering for Detached Generations (#14612)

* 📉 perf: Bound Early Event Buffering for Detached Generations

A generation streaming with no attached subscriber re-entered buffering
mode on every disconnect and retained each emitted event in
earlyEventBuffer for its remaining duration. A single 26-minute detached
run (~58,800 tool-argument deltas) grew the heap past 2 GiB with GC cost
climbing alongside it, while client reconnects always resume from
durable state and discard that local buffer anyway.

- Close the early buffer after the first attachment drains it in Redis
  mode; the durable chunk log and pub/sub own recovery from then on,
  matching how cross-replica subscribers already attach.
- Enforce hard bounds (5,000 events / 8 MB estimated) in both modes; on
  overflow the buffer is discarded and closed, with recovery falling back
  to the durable chunk log (Redis) or resume snapshot (in-memory).
- Add a generation_stream_early_buffer_overflows_total counter and
  earlyBufferedEvents/Bytes gauges on getRuntimeStats() for visibility.
- Add incident-shaped regression tests and update specs that pinned the
  old post-disconnect re-buffering contract.

* fix: redirect post-overflow first attachments to resume recovery

A buffer discarded by the overflow guard left the initial non-resume
SSE attachment with nothing to replay, silently omitting pre-attach
output until the final event. Track the overflow on the runtime and
close such attachments with the existing reconnect signal instead: the
client already re-attaches with resume=true on transport failure and
its sync frame reconstructs the discarded output from durable/snapshot
state. Adds no per-event work; the check is one boolean per attachment.

* fix: enforce buffer bounds when restoring canceled resume captures

Captured emissions restored by a resume canceled before activation
bypassed the early-buffer hard cap, so one oversized restoration could
persist past the limits with no later emission to trip the guard.
Restoration now applies the same overflow-and-close behavior through a
shared helper, and the restore-cap spec fails before this change
(5 events / ~10MB retained) and passes after.

* chore: add Redis management scripts and update package.json for Redis commands

package.json

@@ -40,6 +40,13 @@
     "backend": "cross-env NODE_ENV=production node api/server/index.js",
     "backend:inspect": "cross-env NODE_ENV=production node --inspect --expose-gc api/server/index.js",
     "backend:dev": "cross-env NODE_ENV=development npx nodemon api/server/index.js",
+    "redis:single": "bash scripts/redis-mode.sh single",
+    "redis:cluster": "bash scripts/redis-mode.sh cluster",
+    "redis:stop": "bash scripts/redis-mode.sh stop",
+    "backend:redis:single": "cross-env USE_REDIS=true USE_REDIS_CLUSTER=false REDIS_URI=redis://127.0.0.1:6379 npm run backend",
+    "backend:redis:cluster": "cross-env USE_REDIS=true USE_REDIS_CLUSTER=true REDIS_URI=redis://127.0.0.1:7001,redis://127.0.0.1:7002,redis://127.0.0.1:7003 npm run backend",
+    "backend:dev:redis:single": "cross-env USE_REDIS=true USE_REDIS_CLUSTER=false REDIS_URI=redis://127.0.0.1:6379 npm run backend:dev",
+    "backend:dev:redis:cluster": "cross-env USE_REDIS=true USE_REDIS_CLUSTER=true REDIS_URI=redis://127.0.0.1:7001,redis://127.0.0.1:7002,redis://127.0.0.1:7003 npm run backend:dev",
     "backend:experimental": "cross-env NODE_ENV=production node api/server/experimental.js",
     "backend:stop": "node config/stop-backend.js",
     "build:data-provider": "cd packages/data-provider && npm run build",

packages/api/src/app/metrics.ts

@@ -185,6 +185,7 @@ type GenerationJobMetrics = {
     result: GenerationStreamSubscriptionResult,
   ) => void;
   recordResumePendingEvents: (store: GenerationJobStore, count: number) => void;
+  recordEarlyBufferOverflow: (store: GenerationJobStore) => void;
 };
 
 let generationJobMetrics: GenerationJobMetrics = {
@@ -192,6 +193,7 @@ let generationJobMetrics: GenerationJobMetrics = {
   setJobsInFlight: () => undefined,
   recordSubscription: () => undefined,
   recordResumePendingEvents: () => undefined,
+  recordEarlyBufferOverflow: () => undefined,
 };
 
 type AgentStartupMetrics = {
@@ -241,6 +243,7 @@ const resetMetricRecorders = (): void => {
     setJobsInFlight: () => undefined,
     recordSubscription: () => undefined,
     recordResumePendingEvents: () => undefined,
+    recordEarlyBufferOverflow: () => undefined,
   };
   agentStartupMetrics = {
     recordMilestone: () => undefined,
@@ -277,6 +280,10 @@ export function recordGenerationStreamResumePendingEvents(
   generationJobMetrics.recordResumePendingEvents(store, count);
 }
 
+export function recordGenerationStreamEarlyBufferOverflow(store: GenerationJobStore): void {
+  generationJobMetrics.recordEarlyBufferOverflow(store);
+}
+
 export function recordAgentStartupMilestone(
   milestone: AgentStartupMilestone,
   durationSeconds: number,
@@ -588,6 +595,13 @@ export function createMetrics(): PrometheusMetrics {
     registers: [registry],
   });
 
+  const generationStreamEarlyBufferOverflows = new Counter({
+    name: 'generation_stream_early_buffer_overflows_total',
+    help: 'Early event replay buffers discarded after exceeding hard size bounds',
+    labelNames: ['store'] as const,
+    registers: [registry],
+  });
+
   const agentStartupMilestoneDuration = new Histogram({
     name: 'agent_startup_milestone_duration_seconds',
     help: 'Cumulative agent chat startup latency from request ingress to each milestone',
@@ -632,6 +646,7 @@ export function createMetrics(): PrometheusMetrics {
       generationStreamSubscriptions.inc({ store, type, result }),
     recordResumePendingEvents: (store, count) =>
       generationStreamResumePendingEvents.inc({ store }, count),
+    recordEarlyBufferOverflow: (store) => generationStreamEarlyBufferOverflows.inc({ store }),
   };
 
   agentStartupMetrics = {

packages/api/src/stream/GenerationJobManager.ts

@@ -35,6 +35,13 @@ import type { RecoveredSteerPayload } from './SteerRecovery';
 import type { SteerContentView } from './SteeringLifecycle';
 import type { GenerationJobStore } from '~/app/metrics';
 import type * as t from '~/types';
+import {
+  recordGenerationStreamEarlyBufferOverflow,
+  recordGenerationStreamResumePendingEvents,
+  recordGenerationStreamSubscription,
+  setGenerationJobsInFlight,
+  recordGenerationJob,
+} from '~/app/metrics';
 import {
   JobCreationSupersededError,
   JobPredecessorMismatchError,
@@ -43,12 +50,6 @@ import {
   PAUSE_PERSISTENCE_TIMEOUT_ERROR,
   STEER_QUEUE_MAX_DEPTH,
 } from './interfaces/IJobStore';
-import {
-  recordGenerationStreamResumePendingEvents,
-  recordGenerationStreamSubscription,
-  setGenerationJobsInFlight,
-  recordGenerationJob,
-} from '~/app/metrics';
 import { isRecoveredSteerPayload, RecoveredSteerPayloadMismatchError } from './SteerRecovery';
 import { assertJobStoreV2 } from './jobStoreCapabilities';
 
@@ -99,6 +100,13 @@ export const TERMINAL_PUBLICATION_RECONNECT_ERROR =
  * owner leaves the durable pending bit behind; the next read or subscriber
  * promotes it to conservative reconciliation after this window. */
 const TERMINAL_PERSISTENCE_TIMEOUT_MS = 30_000;
+/** Hard bounds for a runtime's local early-event replay buffer. The buffer
+ * bridges emission to first attachment, but a generation streaming with no
+ * attached subscriber would otherwise grow it for its entire duration. On
+ * overflow the buffer is discarded and closed: Redis mode recovers from the
+ * durable chunk log, in-memory reconnects recover from the resume snapshot. */
+const EARLY_EVENT_BUFFER_MAX_EVENTS = 5_000;
+const EARLY_EVENT_BUFFER_MAX_BYTES = 8 * 1024 * 1024;
 const CLIENT_REQUEST_ID_PATTERN = /^[A-Za-z0-9:_-]{1,128}$/;
 type TokenIdempotencyClaim = IdempotencyClaimValue & {
   claimedAt: number;
@@ -523,6 +531,16 @@ interface RuntimeJobState {
   pausePersistenceTimeoutRetired?: boolean;
   syncSent: boolean;
   earlyEventBuffer: t.ServerSentEvent[];
+  /** Estimated serialized size of earlyEventBuffer, for the overflow guard. */
+  earlyEventBufferBytes: number;
+  /** Closed after the first attachment drains the buffer in Redis mode (the
+   * durable chunk log owns later recovery) or after an overflow discard. A
+   * closed buffer never re-accumulates events for this runtime. */
+  earlyEventBufferClosed: boolean;
+  /** The buffer was discarded by the overflow guard before a subscriber
+   * consumed it. Non-resume attachments are redirected to the resume path,
+   * which reconstructs the discarded output from durable/snapshot state. */
+  earlyEventBufferOverflowed?: true;
   earlyEventSequencePromises: Array<Promise<void | number>>;
   /** Initial subscribers eligible to receive the local pre-attachment replay. */
   earlyReplayHandlers: Set<t.ChunkHandler>;
@@ -2148,6 +2166,8 @@ class GenerationJobManagerClass {
       startupTelemetry: options.startupTelemetry,
       syncSent: false,
       earlyEventBuffer: [],
+      earlyEventBufferBytes: 0,
+      earlyEventBufferClosed: false,
       earlyEventSequencePromises: [],
       earlyReplayHandlers: new Set(),
       resumeCaptureHandlers: new Set(),
@@ -2406,6 +2426,8 @@ class GenerationJobManagerClass {
       resolveReady,
       syncSent: jobData.syncSent ?? false,
       earlyEventBuffer: [],
+      earlyEventBufferBytes: 0,
+      earlyEventBufferClosed: false,
       earlyEventSequencePromises: [],
       earlyReplayHandlers: new Set(),
       resumeCaptureHandlers: new Set(),
@@ -4185,8 +4207,14 @@ class GenerationJobManagerClass {
           }
         }
       } finally {
-        runtime.earlyEventBuffer = [];
-        runtime.earlyEventSequencePromises = [];
+        this.resetEarlyEventBuffer(runtime);
+        if (this._isRedis) {
+          /** After the first attachment, the durable chunk log and pub/sub own
+           * recovery; re-buffering on a later detach would grow for the rest
+           * of a detached run. Cross-replica subscribers already attach with
+           * no local buffer, so a closed buffer follows the same path. */
+          runtime.earlyEventBufferClosed = true;
+        }
         try {
           const reorderSync = this.eventTransport.syncReorderBuffer?.(streamId, replayedNextSeq);
           if (reorderSync) {
@@ -4212,6 +4240,20 @@ class GenerationJobManagerClass {
       return null;
     }
 
+    if (
+      runtime.earlyEventBufferOverflowed === true &&
+      !options?.skipBufferReplay &&
+      !runtime.finalEvent &&
+      !runtime.errorEvent
+    ) {
+      /** The overflow guard discarded the pre-attachment buffer, so a
+       * non-resume attachment cannot be made whole from local replay. Close
+       * the transport with the reconnect signal instead of streaming a
+       * silently truncated response: the client re-attaches with resume=true
+       * and its sync frame carries full durable/snapshot state. */
+      queueError(TERMINAL_PUBLICATION_RECONNECT_ERROR);
+    }
+
     recordGenerationStreamSubscription(this.storeLabel, subscriptionType, 'success');
 
     if (isFirst) {
@@ -4361,6 +4403,58 @@ class GenerationJobManagerClass {
     await Promise.all(pending);
   }
 
+  private resetEarlyEventBuffer(runtime: RuntimeJobState): void {
+    runtime.earlyEventBuffer = [];
+    runtime.earlyEventSequencePromises = [];
+    runtime.earlyEventBufferBytes = 0;
+  }
+
+  /**
+   * Buffers a pre-attachment event for local replay, enforcing hard bounds.
+   *
+   * A generation streaming with no attached subscriber can run for its entire
+   * duration; unbounded buffering here retained every emitted event in memory,
+   * with GC cost climbing alongside the heap. On overflow the whole buffer is
+   * discarded and closed — the durable chunk log (Redis) or the resume
+   * snapshot (in-memory) already owns recovery for late subscribers.
+   *
+   * @returns whether the event was accepted into the buffer.
+   */
+  private bufferEarlyEvent(
+    streamId: string,
+    runtime: RuntimeJobState,
+    event: t.ServerSentEvent,
+  ): boolean {
+    if (runtime.earlyEventBufferClosed) {
+      return false;
+    }
+    const estimatedBytes = JSON.stringify(event).length;
+    if (
+      runtime.earlyEventBuffer.length >= EARLY_EVENT_BUFFER_MAX_EVENTS ||
+      runtime.earlyEventBufferBytes + estimatedBytes > EARLY_EVENT_BUFFER_MAX_BYTES
+    ) {
+      this.overflowEarlyEventBuffer(streamId, runtime);
+      return false;
+    }
+    runtime.earlyEventBuffer.push(event);
+    runtime.earlyEventBufferBytes += estimatedBytes;
+    return true;
+  }
+
+  private overflowEarlyEventBuffer(streamId: string, runtime: RuntimeJobState): void {
+    const droppedEvents = runtime.earlyEventBuffer.length;
+    const droppedBytes = runtime.earlyEventBufferBytes;
+    this.resetEarlyEventBuffer(runtime);
+    runtime.earlyEventBufferClosed = true;
+    runtime.earlyEventBufferOverflowed = true;
+    recordGenerationStreamEarlyBufferOverflow(this.storeLabel);
+    logger.warn(
+      `[GenerationJobManager] Early event buffer overflow for ${streamId}; ` +
+        `discarded ${droppedEvents} buffered events (~${droppedBytes} bytes); ` +
+        'late subscribers will recover from durable/resume state',
+    );
+  }
+
   /**
    * If the subscriber that owns Redis attachment bootstrap disconnects, finish the
    * replay/sync for any concurrent subscriber. Otherwise the transport-wide reorder
@@ -4430,8 +4524,11 @@ class GenerationJobManagerClass {
             }
           }
         } finally {
-          runtime.earlyEventBuffer = [];
-          runtime.earlyEventSequencePromises = [];
+          this.resetEarlyEventBuffer(runtime);
+          if (this._isRedis) {
+            /** Same closure as the owning-subscriber bootstrap above. */
+            runtime.earlyEventBufferClosed = true;
+          }
           await this.eventTransport.syncReorderBuffer?.(streamId, replayedNextSeq);
         }
       })
@@ -4522,11 +4619,31 @@ class GenerationJobManagerClass {
         return;
       }
       const currentRuntime = this.runtimeState.get(streamId);
-      if (currentRuntime && !currentRuntime.hasSubscriber) {
+      if (
+        currentRuntime &&
+        !currentRuntime.hasSubscriber &&
+        !currentRuntime.earlyEventBufferClosed
+      ) {
         const bufferedEvents = new Set(currentRuntime.earlyEventBuffer);
         const missingEvents = capturedPendingEvents.filter((event) => !bufferedEvents.has(event));
         if (missingEvents.length > 0) {
-          currentRuntime.earlyEventBuffer = [...missingEvents, ...currentRuntime.earlyEventBuffer];
+          let restoredBytes = 0;
+          for (const event of missingEvents) {
+            restoredBytes += JSON.stringify(event).length;
+          }
+          const overflows =
+            currentRuntime.earlyEventBuffer.length + missingEvents.length >
+              EARLY_EVENT_BUFFER_MAX_EVENTS ||
+            currentRuntime.earlyEventBufferBytes + restoredBytes > EARLY_EVENT_BUFFER_MAX_BYTES;
+          if (overflows) {
+            this.overflowEarlyEventBuffer(streamId, currentRuntime);
+          } else {
+            currentRuntime.earlyEventBuffer = [
+              ...missingEvents,
+              ...currentRuntime.earlyEventBuffer,
+            ];
+            currentRuntime.earlyEventBufferBytes += restoredBytes;
+          }
         }
       }
       capturedPendingEvents.length = 0;
@@ -5111,15 +5228,13 @@ class GenerationJobManagerClass {
       }
     }
 
-    const buffered = !runtime.hasSubscriber;
-    if (buffered) {
-      runtime.earlyEventBuffer.push(event);
-      if (!this._isRedis) {
-        if (runtime.startupTelemetry) {
-          this.recordStartupEvent(runtime, event);
-        }
-        return;
+    const detached = !runtime.hasSubscriber;
+    const buffered = detached && this.bufferEarlyEvent(streamId, runtime, event);
+    if (detached && !this._isRedis) {
+      if (runtime.startupTelemetry) {
+        this.recordStartupEvent(runtime, event);
       }
+      return;
     }
 
     if (!buffered && !runtime.startupTelemetry) {
@@ -6474,11 +6589,21 @@ class GenerationJobManagerClass {
     runtimeStateSize: number;
     runStepBufferSize: number;
     eventTransportStreams: number;
+    earlyBufferedEvents: number;
+    earlyBufferedBytes: number;
   } {
+    let earlyBufferedEvents = 0;
+    let earlyBufferedBytes = 0;
+    for (const runtime of this.runtimeState.values()) {
+      earlyBufferedEvents += runtime.earlyEventBuffer.length;
+      earlyBufferedBytes += runtime.earlyEventBufferBytes;
+    }
     return {
       runtimeStateSize: this.runtimeState.size,
       runStepBufferSize: this.runStepBuffers?.size ?? 0,
       eventTransportStreams: this.eventTransport.getTrackedStreamIds().length,
+      earlyBufferedEvents,
+      earlyBufferedBytes,
     };
   }
 

packages/api/src/stream/__tests__/GenerationJobManager.stream_integration.spec.ts

@@ -6,11 +6,14 @@ import {
   keyvRedisClient as staticKeyvClient,
   keyvRedisClientReady,
 } from '~/cache/redisClients';
+import {
+  GenerationJobManagerClass,
+  TERMINAL_PUBLICATION_RECONNECT_ERROR,
+} from '~/stream/GenerationJobManager';
 import { InMemoryEventTransport } from '~/stream/implementations/InMemoryEventTransport';
 import { RedisEventTransport } from '~/stream/implementations/RedisEventTransport';
 import { InMemoryJobStore } from '~/stream/implementations/InMemoryJobStore';
 import { STEER_ENQUEUE_NOT_RUNNING } from '~/stream/interfaces/IJobStore';
-import { GenerationJobManagerClass } from '~/stream/GenerationJobManager';
 import { RedisJobStore } from '~/stream/implementations/RedisJobStore';
 import { createStreamServices } from '~/stream/createStreamServices';
 import { GenerationJobManager } from '~/stream/GenerationJobManager';
@@ -1588,11 +1591,77 @@ describe('GenerationJobManager Integration Tests', () => {
       await manager.destroy();
     });
 
+    testRedis('should not re-buffer detached events after first attachment (Redis)', async () => {
+      /**
+       * After the first attachment, the durable chunk log owns recovery for
+       * detached events; re-buffering them locally grew without bound for
+       * long detached generations. A late subscriber gets live events only,
+       * and resume reconstructs the detached content from the chunk log.
+       */
+      const manager = createRedisManager();
+      const streamId = `no-rebuffer-redis-${Date.now()}`;
+      await manager.createJob(streamId, 'user-1');
+
+      const sub1 = await manager.subscribe(streamId, () => {});
+      await new Promise((resolve) => setTimeout(resolve, 100));
+      await manager.emitChunk(streamId, {
+        event: 'on_run_step',
+        data: {
+          id: 'step-1',
+          runId: 'run-1',
+          index: 0,
+          stepDetails: { type: 'message_creation' },
+        },
+      });
+      await new Promise((resolve) => setTimeout(resolve, 100));
+      sub1?.unsubscribe();
+      await new Promise((resolve) => setTimeout(resolve, 100));
+
+      await manager.emitChunk(streamId, {
+        event: 'on_message_delta',
+        data: { id: 'step-1', delta: { content: { type: 'text', text: 'detached-redis' } } },
+      });
+
+      await new Promise((resolve) => setTimeout(resolve, 100));
+      expect(manager.getRuntimeStats().earlyBufferedEvents).toBe(0);
+
+      const sub2Events: ServerSentEvent[] = [];
+      const sub2 = await manager.subscribe(streamId, (event) => sub2Events.push(event));
+
+      await new Promise((resolve) => setTimeout(resolve, 200));
+      expect(sub2Events.length).toBe(0);
+
+      const resumeState = await manager.getResumeState(streamId);
+      expect(JSON.stringify(resumeState?.aggregatedContent ?? [])).toContain('detached-redis');
+
+      await manager.emitChunk(streamId, {
+        event: 'on_message_delta',
+        data: { id: 'step-1', delta: { content: { type: 'text', text: ' live' } } },
+      });
+
+      await new Promise((resolve) => setTimeout(resolve, 200));
+      expect(sub2Events.length).toBe(1);
+      expect((sub2Events[0] as StreamEvent).event).toBe('on_message_delta');
+
+      sub2?.unsubscribe();
+      await manager.destroy();
+    });
+  });
+
+  describe('Early event buffer bounds', () => {
+    /**
+     * Regression tests for a production incident: a model streamed a malformed
+     * 150k-character tool argument for ~26 minutes after the browser
+     * disconnected (~40 events/sec, ~58,800 publications). Every event was
+     * retained in earlyEventBuffer, so heap and GC cost climbed for the whole
+     * detached run.
+     */
+
     testRedis(
-      'should replay buffer without skipBufferReplay after disconnect (Redis)',
+      'detached generation keeps the local buffer empty after first attachment (Redis)',
       async () => {
         const manager = createRedisManager();
-        const streamId = `replay-buf-redis-${Date.now()}`;
+        const streamId = `detached-flat-${Date.now()}`;
         await manager.createJob(streamId, 'user-1');
 
         const sub1 = await manager.subscribe(streamId, () => {});
@@ -1600,25 +1669,204 @@ describe('GenerationJobManager Integration Tests', () => {
         sub1?.unsubscribe();
         await new Promise((resolve) => setTimeout(resolve, 100));
 
-        await manager.emitChunk(streamId, {
-          event: 'on_message_delta',
-          data: { delta: { content: { type: 'text', text: 'buffered-redis' } } },
-        });
+        for (let i = 0; i < 200; i++) {
+          await manager.emitChunk(streamId, {
+            event: 'on_run_step_delta',
+            data: {
+              id: 'step-1',
+              delta: { type: 'tool_call_delta', args: `"table_${i}", ` },
+            },
+          });
+        }
 
-        await new Promise((resolve) => setTimeout(resolve, 100));
+        const stats = manager.getRuntimeStats();
+        expect(stats.earlyBufferedEvents).toBe(0);
+        expect(stats.earlyBufferedBytes).toBe(0);
 
-        const sub2Events: ServerSentEvent[] = [];
-        const sub2 = await manager.subscribe(streamId, (event) => sub2Events.push(event));
-
-        await new Promise((resolve) => setTimeout(resolve, 200));
-
-        expect(sub2Events.length).toBe(1);
-        expect((sub2Events[0] as StreamEvent).event).toBe('on_message_delta');
-
-        sub2?.unsubscribe();
         await manager.destroy();
       },
     );
+
+    test('discards and closes the buffer when the byte budget is exceeded (never attached)', async () => {
+      const manager = createInMemoryManager();
+      const streamId = `buf-byte-cap-${Date.now()}`;
+      await manager.createJob(streamId, 'user-1');
+
+      const bigText = 'x'.repeat(2 * 1024 * 1024);
+      for (let i = 0; i < 5; i++) {
+        await manager.emitChunk(streamId, {
+          event: 'on_message_delta',
+          data: { id: 'step-1', delta: { content: { type: 'text', text: bigText } } },
+        });
+      }
+
+      const stats = manager.getRuntimeStats();
+      expect(stats.earlyBufferedEvents).toBe(0);
+      expect(stats.earlyBufferedBytes).toBe(0);
+
+      await manager.emitChunk(streamId, {
+        event: 'on_message_delta',
+        data: { id: 'step-1', delta: { content: { type: 'text', text: 'after-overflow' } } },
+      });
+      expect(manager.getRuntimeStats().earlyBufferedEvents).toBe(0);
+
+      const errors: string[] = [];
+      const events: ServerSentEvent[] = [];
+      const sub = await manager.subscribe(
+        streamId,
+        (event) => events.push(event),
+        undefined,
+        (error) => errors.push(error),
+      );
+      await new Promise((resolve) => setTimeout(resolve, 20));
+
+      /** A non-resume attachment cannot be made whole once the buffer was
+       * discarded, so it is closed with the reconnect signal; the client then
+       * re-attaches with resume=true and syncs from snapshot state. */
+      expect(errors).toEqual([TERMINAL_PUBLICATION_RECONNECT_ERROR]);
+      expect(events).toEqual([]);
+
+      sub?.unsubscribe();
+      await manager.destroy();
+    });
+
+    testRedis('redirects a post-overflow first attachment to resume recovery (Redis)', async () => {
+      const manager = createRedisManager();
+      const streamId = `overflow-redirect-${Date.now()}`;
+      await manager.createJob(streamId, 'user-1');
+
+      await manager.emitChunk(streamId, {
+        event: 'on_run_step',
+        data: {
+          id: 'step-1',
+          runId: 'run-1',
+          index: 0,
+          stepDetails: { type: 'message_creation' },
+        },
+      });
+      const bigText = 'y'.repeat(2 * 1024 * 1024);
+      for (let i = 0; i < 5; i++) {
+        await manager.emitChunk(streamId, {
+          event: 'on_message_delta',
+          data: { id: 'step-1', delta: { content: { type: 'text', text: bigText } } },
+        });
+      }
+      expect(manager.getRuntimeStats().earlyBufferedEvents).toBe(0);
+
+      const errors: string[] = [];
+      const sub = await manager.subscribe(
+        streamId,
+        () => {},
+        undefined,
+        (error) => errors.push(error),
+      );
+      await new Promise((resolve) => setTimeout(resolve, 100));
+      expect(errors).toEqual([TERMINAL_PUBLICATION_RECONNECT_ERROR]);
+      sub?.unsubscribe();
+      await new Promise((resolve) => setTimeout(resolve, 100));
+
+      /** The resume path the client falls back to reconstructs the
+       * discarded output from the durable chunk log. */
+      const resumeState = await manager.getResumeState(streamId);
+      expect(JSON.stringify(resumeState?.aggregatedContent ?? [])).toContain('yyyy');
+
+      await manager.destroy();
+    });
+
+    test('buffers detached events until the cap in in-memory mode', async () => {
+      const manager = createInMemoryManager();
+      const streamId = `buf-below-cap-${Date.now()}`;
+      await manager.createJob(streamId, 'user-1');
+
+      await setupDisconnectedStream(manager, streamId, 10);
+
+      const stats = manager.getRuntimeStats();
+      expect(stats.earlyBufferedEvents).toBe(2);
+      expect(stats.earlyBufferedBytes).toBeGreaterThan(0);
+
+      await manager.destroy();
+    });
+
+    test('caps captured events restored by a resume canceled before activation', async () => {
+      const jobStore = new InMemoryJobStore({ ttlAfterComplete: 60000 });
+      const manager = new GenerationJobManagerClass();
+      manager.configure({
+        jobStore,
+        eventTransport: new InMemoryEventTransport(),
+        isRedis: false,
+      });
+      manager.initialize();
+      const streamId = `restore-cap-${Date.now()}`;
+      await manager.createJob(streamId, 'user-1');
+
+      await setupDisconnectedStream(manager, streamId, 10);
+
+      /** Arm only after the snapshot completes so the gate parks the resume
+       * in its post-attachment steer reconciliation, the window where
+       * emissions are captured per-resume instead of buffered. */
+      let armed = false;
+      const originalGetResumeState = manager.getResumeState.bind(manager);
+      jest
+        .spyOn(manager, 'getResumeState')
+        .mockImplementation(
+          async (...args: Parameters<GenerationJobManagerClass['getResumeState']>) => {
+            const result = await originalGetResumeState(...args);
+            armed = true;
+            return result;
+          },
+        );
+      let releaseGate!: () => void;
+      const gate = new Promise<void>((resolve) => (releaseGate = resolve));
+      let gateReached!: () => void;
+      const reached = new Promise<void>((resolve) => (gateReached = resolve));
+      const originalPeek = jobStore.peekSteers.bind(jobStore);
+      let gated = true;
+      jest
+        .spyOn(jobStore, 'peekSteers')
+        .mockImplementation(async (...args: Parameters<InMemoryJobStore['peekSteers']>) => {
+          if (armed && gated) {
+            gated = false;
+            gateReached();
+            await gate;
+          }
+          return originalPeek(...args);
+        });
+
+      const resumePromise = manager.subscribeWithResume(streamId, () => {});
+
+      await reached;
+      const bigText = 'z'.repeat(2 * 1024 * 1024);
+      for (let i = 0; i < 5; i++) {
+        await manager.emitChunk(streamId, {
+          event: 'on_message_delta',
+          data: { id: 'step-1', delta: { content: { type: 'text', text: bigText } } },
+        });
+      }
+      releaseGate();
+
+      const { subscription } = await resumePromise;
+      expect(subscription).not.toBeNull();
+      subscription!.unsubscribe();
+
+      /** The ~10MB of captured events must not survive restoration. */
+      const stats = manager.getRuntimeStats();
+      expect(stats.earlyBufferedEvents).toBe(0);
+      expect(stats.earlyBufferedBytes).toBe(0);
+
+      /** Restoration overflowed, so a non-resume attach takes the redirect. */
+      const errors: string[] = [];
+      const probe = await manager.subscribe(
+        streamId,
+        () => {},
+        undefined,
+        (error) => errors.push(error),
+      );
+      await new Promise((resolve) => setTimeout(resolve, 20));
+      expect(errors).toEqual([TERMINAL_PUBLICATION_RECONNECT_ERROR]);
+      probe?.unsubscribe();
+
+      await manager.destroy();
+    });
   });
 
   describe('Atomic subscribeWithResume', () => {

packages/api/src/stream/__tests__/RedisEventTransport.spec.ts

@@ -1557,10 +1557,14 @@ describe('RedisEventTransport', () => {
 
     subscription?.unsubscribe();
 
+    /** After the first attachment the local buffer stays closed; a detached
+     * emission is durable-log-only, and the reconnect frontier must advance
+     * past its sequence so the next live chunk is not held for reordering. */
     await manager.emitChunk(streamId, {
       event: 'on_message_delta',
-      data: { delta: { content: { type: 'text', text: 'buffered after disconnect' } } },
+      data: { delta: { content: { type: 'text', text: 'detached after disconnect' } } },
     });
+    expect(manager.getRuntimeStats().earlyBufferedEvents).toBe(0);
 
     const resumed: unknown[] = [];
     const resumedSubscription = await manager.subscribe(streamId, (event) => resumed.push(event));
@@ -1570,10 +1574,6 @@ describe('RedisEventTransport', () => {
     });
 
     expect(resumed).toEqual([
-      {
-        event: 'on_message_delta',
-        data: { delta: { content: { type: 'text', text: 'buffered after disconnect' } } },
-      },
       {
         event: 'on_message_delta',
         data: { delta: { content: { type: 'text', text: 'live after reconnect' } } },

packages/api/src/stream/__tests__/reconnect-reorder-desync.stream_integration.spec.ts

@@ -867,7 +867,7 @@ describe('Reconnect Reorder Buffer Desync (Regression)', () => {
       await manager.destroy();
     });
 
-    test('mid-generation buffer replay advances to its absolute Redis sequence', async () => {
+    test('early buffer replay advances to its absolute Redis sequence', async () => {
       if (!ioredisClient) {
         console.warn('Redis not available, skipping test');
         return;
@@ -885,24 +885,17 @@ describe('Reconnect Reorder Buffer Desync (Regression)', () => {
       const streamId = `absolute-replay-${Date.now()}`;
       await manager.createJob(streamId, 'user-1');
 
-      const firstEvents: unknown[] = [];
-      const firstSubscription = await manager.subscribe(streamId, (event) => {
-        firstEvents.push(event);
-      });
-      await new Promise((resolve) => setTimeout(resolve, 100));
+      // Redis sequences are conversation-scoped and survive across turns, so a
+      // fresh generation's early buffer can hold events whose sequences start
+      // well above zero. Seed the shared counter to model that later turn.
+      await ioredisClient.set(`stream:{${streamId}}:seq`, '5');
 
+      // These buffered events receive seqs 5 and 6. Replaying two events must
+      // therefore advance to seq=7, not to the relative count of 2.
       await manager.emitChunk(streamId, {
         event: 'on_message_delta',
         data: { index: 0 },
       });
-      await new Promise((resolve) => setTimeout(resolve, 100));
-      expect(firstEvents).toHaveLength(1);
-
-      firstSubscription?.unsubscribe();
-      await new Promise((resolve) => setTimeout(resolve, 100));
-
-      // This buffered event receives seq=1. Replaying one event must therefore
-      // advance to seq=2, not to the relative count of 1.
       await manager.emitChunk(streamId, {
         event: 'on_message_delta',
         data: { index: 1 },
@@ -923,7 +916,7 @@ describe('Reconnect Reorder Buffer Desync (Regression)', () => {
 
       expect(
         resumedEvents.map((event) => (event as { data: { index: number } }).data.index),
-      ).toEqual([1, 2]);
+      ).toEqual([0, 1, 2]);
 
       resumedSubscription?.unsubscribe();
       await manager.destroy();

scripts/redis-mode.sh

@@ -0,0 +1,60 @@
+#!/usr/bin/env bash
+
+set -euo pipefail
+
+ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
+CLUSTER_DIR="$ROOT_DIR/redis-config"
+
+require_redis() {
+  if command -v redis-server >/dev/null && command -v redis-cli >/dev/null; then
+    return
+  fi
+
+  echo "Redis is required. Install it with: sudo apt-get install redis-server redis-tools"
+  exit 1
+}
+
+redis_is_running() {
+  redis-cli -p "$1" ping >/dev/null 2>&1
+}
+
+stop_single() {
+  if redis_is_running 6379; then
+    redis-cli -p 6379 shutdown nosave >/dev/null 2>&1 || true
+  fi
+}
+
+start_single() {
+  mkdir -p "$CLUSTER_DIR/data/6379"
+
+  if redis_is_running 6379; then
+    echo "Redis single node is already running on port 6379."
+    return
+  fi
+
+  redis-server --port 6379 --dir "$CLUSTER_DIR/data/6379" --save '' --appendonly no --daemonize yes
+  redis-cli -p 6379 ping >/dev/null
+  echo "Redis single node is ready on port 6379."
+}
+
+case "${1:-}" in
+  single)
+    require_redis
+    "$CLUSTER_DIR/stop-cluster.sh" >/dev/null 2>&1 || true
+    start_single
+    ;;
+  cluster)
+    require_redis
+    stop_single
+    exec "$CLUSTER_DIR/start-cluster.sh"
+    ;;
+  stop)
+    require_redis
+    stop_single
+    exec "$CLUSTER_DIR/stop-cluster.sh"
+    ;;
+  *)
+    echo "Usage: $0 {single|cluster|stop}"
+    exit 1
+    ;;
+esac