♾️ fix: Preserve Resumable Stream Ordering Across Turns (#14411)

* fix: preserve resumable stream ordering across turns

* chore: sort stream regression imports

* test: mirror sliding sequence ttl in publisher mock

* fix: prevent duplicate early stream replay

* fix: preserve replay frontier when sync fails

packages/api/src/stream/GenerationJobManager.ts

@@ -152,6 +152,8 @@ export interface GenerationJobManagerOptions {
  * @property errorEvent - Cached error event for late subscribers (errors before client connects)
  * @property syncSent - Whether sync event was sent (reset when all subscribers leave)
  * @property earlyEventBuffer - Buffer for events emitted before first subscriber connects
+ * @property earlyEventSequencePromises - Redis sequence assignments corresponding to buffered
+ *   events. Their absolute values identify the exact ordering frontier after replay.
  * @property hasSubscriber - Whether at least one subscriber has connected
  * @property allSubscribersLeftHandlers - Internal handlers for disconnect events.
  *   These are stored separately from eventTransport subscribers to avoid being counted
@@ -169,6 +171,7 @@ interface RuntimeJobState {
   approvalCleanupRan?: boolean;
   syncSent: boolean;
   earlyEventBuffer: t.ServerSentEvent[];
+  earlyEventSequencePromises: Array<Promise<void | number>>;
   hasSubscriber: boolean;
   allSubscribersLeftHandlers?: Array<(...args: unknown[]) => void>;
 }
@@ -391,6 +394,7 @@ class GenerationJobManagerClass {
       resolveReady: resolveReady!,
       syncSent: false,
       earlyEventBuffer: [],
+      earlyEventSequencePromises: [],
       hasSubscriber: false,
     };
     this.runtimeState.set(streamId, runtime);
@@ -613,6 +617,7 @@ class GenerationJobManagerClass {
       resolveReady: resolveReady!,
       syncSent: jobData.syncSent ?? false,
       earlyEventBuffer: [],
+      earlyEventSequencePromises: [],
       hasSubscriber: false,
       finalEvent,
       errorEvent: jobData.error,
@@ -1048,16 +1053,26 @@ class GenerationJobManagerClass {
       }
     });
 
-    const subscription = this.eventTransport.subscribe(streamId, {
-      onChunk: (event) => {
-        const e = event as t.ServerSentEvent;
-        if (!(e as Record<string, unknown>)._internal) {
-          onChunk(e);
-        }
+    const subscription = this.eventTransport.subscribe(
+      streamId,
+      {
+        onChunk: (event) => {
+          const e = event as t.ServerSentEvent;
+          if (!(e as Record<string, unknown>)._internal) {
+            onChunk(e);
+          }
+        },
+        onDone: (event) => onDone?.(event as t.ServerSentEvent),
+        onError,
       },
-      onDone: (event) => onDone?.(event as t.ServerSentEvent),
-      onError,
-    });
+      {
+        // Redis can publish an early buffered event before the EVAL response carrying its
+        // sequence reaches this process. Hold sequenced pub/sub delivery until replay and
+        // sync establish the exact frontier, otherwise the new subscriber sees it twice.
+        deferSequenceDelivery:
+          this._isRedis && !runtime.hasSubscriber && !options?.skipBufferReplay,
+      },
+    );
 
     try {
       if (subscription.ready) {
@@ -1075,32 +1090,40 @@ class GenerationJobManagerClass {
       runtime.hasSubscriber = true;
 
       /**
-       * Pass earlyReplayCount to syncReorderBuffer so it can prune duplicate pub/sub
-       * entries (seqs 0..count-1) without touching live in-flight chunks.
+       * The Redis sequence is conversation-scoped and therefore may start this
+       * generation above zero. Synchronize with the absolute sequence frontier
+       * assigned to the events replayed below, never with their relative count.
        *
-       * Only set when the buffer was actually replayed — those specific seqs were
-       * delivered via onChunk and their pub/sub copies are duplicates.
        * When skipBufferReplay is true, the resume sync payload delivers aggregated
-       * content up to the Redis counter, so syncReorderBuffer should trust currentSeq
-       * as the frontier (earlyReplayCount = 0).
+       * content up to the Redis counter, so syncReorderBuffer receives no local
+       * replay frontier and trusts the current counter.
        */
-      let earlyReplayCount = 0;
+      let replayedNextSeq: number | undefined;
+      const bufferedEvents = runtime.earlyEventBuffer;
+      const sequencePromises = runtime.earlyEventSequencePromises;
+      runtime.earlyEventBuffer = [];
+      runtime.earlyEventSequencePromises = [];
 
-      if (runtime.earlyEventBuffer.length > 0) {
+      if (bufferedEvents.length > 0) {
+        const sequences = await Promise.all(sequencePromises);
         if (options?.skipBufferReplay) {
           logger.debug(
-            `[GenerationJobManager] Skipping ${runtime.earlyEventBuffer.length} buffered events for ${streamId} (skipBufferReplay)`,
+            `[GenerationJobManager] Skipping ${bufferedEvents.length} buffered events for ${streamId} (skipBufferReplay)`,
           );
         } else {
-          earlyReplayCount = runtime.earlyEventBuffer.length;
-          logger.debug(
-            `[GenerationJobManager] Replaying ${earlyReplayCount} buffered events for ${streamId}`,
+          const assignedSequences = sequences.filter(
+            (sequence): sequence is number => typeof sequence === 'number',
           );
-          for (const bufferedEvent of runtime.earlyEventBuffer) {
+          if (assignedSequences.length > 0) {
+            replayedNextSeq = Math.max(...assignedSequences) + 1;
+          }
+          logger.debug(
+            `[GenerationJobManager] Replaying ${bufferedEvents.length} buffered events for ${streamId}`,
+          );
+          for (const bufferedEvent of bufferedEvents) {
             onChunk(bufferedEvent);
           }
         }
-        runtime.earlyEventBuffer = [];
       } else if (this._isRedis && !options?.skipBufferReplay && jobData?.userMessage) {
         /**
          * Cross-replica fallback: the created event was buffered on the generating
@@ -1125,7 +1148,7 @@ class GenerationJobManagerClass {
       }
 
       try {
-        await this.eventTransport.syncReorderBuffer?.(streamId, earlyReplayCount);
+        await this.eventTransport.syncReorderBuffer?.(streamId, replayedNextSeq);
       } catch (err) {
         logger.warn(
           `[GenerationJobManager] Failed to sync reorder buffer for ${streamId}; proceeding with current nextSeq:`,
@@ -1324,14 +1347,21 @@ class GenerationJobManagerClass {
       }
     }
 
-    if (!runtime.hasSubscriber) {
+    const shouldBuffer = !runtime.hasSubscriber;
+    if (shouldBuffer) {
       runtime.earlyEventBuffer.push(event);
       if (!this._isRedis) {
         return;
       }
     }
 
-    await this.eventTransport.emitChunk(streamId, event);
+    const publishPromise = Promise.resolve(this.eventTransport.emitChunk(streamId, event));
+    if (shouldBuffer) {
+      // Store the promise before yielding so subscribe() can wait for the exact
+      // sequence assignment that belongs to every event it replays.
+      runtime.earlyEventSequencePromises.push(publishPromise);
+    }
+    await publishPromise;
   }
 
   /**
@@ -2030,17 +2060,25 @@ class GenerationJobManagerClass {
     let runningJobsChanged = false;
 
     // Cleanup runtime state for deleted jobs
-    for (const streamId of this.runtimeState.keys()) {
+    for (const [streamId, observedRuntime] of this.runtimeState) {
       if (!(await this.jobStore.hasJob(streamId))) {
+        // A replacement generation can reuse the same streamId while hasJob()
+        // is in flight. Never reap the replacement runtime based on the stale
+        // absence observed for its predecessor.
+        if (this.runtimeState.get(streamId) !== observedRuntime) {
+          if (!observedRuntime.abortController.signal.aborted) {
+            observedRuntime.abortController.abort();
+          }
+          continue;
+        }
         /**
          * Abort any still-pending generation whose job has been reaped (e.g. a
          * stale "running" job removed by the store's failsafe timeout). This
          * unwinds the hung in-flight work so its client/graph references can be
          * garbage collected, rather than leaking via the pending promise.
          */
-        const runtime = this.runtimeState.get(streamId);
-        if (runtime && !runtime.abortController.signal.aborted) {
-          runtime.abortController.abort();
+        if (!observedRuntime.abortController.signal.aborted) {
+          observedRuntime.abortController.abort();
         }
         // If a client is still attached when the job is reaped, send a terminal
         // error first so the SSE connection closes instead of hanging open with no
@@ -2052,6 +2090,11 @@ class GenerationJobManagerClass {
             logger.error(`[GenerationJobManager] Failed to notify reaped stream ${streamId}:`, err);
           }
         }
+        // emitError() is asynchronous; a replacement may have appeared while
+        // the terminal event was being published.
+        if (this.runtimeState.get(streamId) !== observedRuntime) {
+          continue;
+        }
         this.runtimeState.delete(streamId);
         runningJobsChanged = this.runningJobs.delete(streamId) || runningJobsChanged;
         this.runStepBuffers?.delete(streamId);

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

@@ -226,7 +226,7 @@ describe('RedisEventTransport Integration Tests', () => {
       subscriber.disconnect();
     });
 
-    test('should assign 0-indexed sequences and set a TTL on the counter only once', async () => {
+    test('should assign 0-indexed sequences and refresh a shortened counter TTL', async () => {
       if (!ioredisClient) {
         console.warn('Redis not available, skipping test');
         return;
@@ -246,22 +246,53 @@ describe('RedisEventTransport Integration Tests', () => {
       await new Promise((resolve) => setTimeout(resolve, 100));
 
       await transport.emitChunk(streamId, { index: 0 });
-      /** First INCR arms the TTL; it must never be refreshed, or a long stream could
-       *  have its counter reset mid-generation. */
+      /** First INCR arms the safety TTL. */
       const ttlAfterFirst = await (ioredisClient as Redis).ttl(seqKey);
       expect(ttlAfterFirst).toBeGreaterThan(0);
 
-      for (let i = 1; i < 5; i++) {
+      // Simulate a nearly-expired counter. The next publish must restore the
+      // safety window so a live generation cannot reset to sequence zero.
+      await (ioredisClient as Redis).expire(seqKey, 2);
+      await transport.emitChunk(streamId, { index: 1 });
+      expect(await (ioredisClient as Redis).ttl(seqKey)).toBeGreaterThan(86_000);
+
+      for (let i = 2; i < 5; i++) {
         await transport.emitChunk(streamId, { index: i });
       }
 
       /** Counter is 1-based in Redis; seq is 0-based, so 5 emits => counter 5, last seq 4. */
       expect(await (ioredisClient as Redis).get(seqKey)).toBe('5');
-      expect(await (ioredisClient as Redis).ttl(seqKey)).toBeLessThanOrEqual(ttlAfterFirst);
+      expect(await (ioredisClient as Redis).ttl(seqKey)).toBeGreaterThanOrEqual(ttlAfterFirst - 1);
 
       transport.destroy();
       subscriber.disconnect();
     });
+
+    test('mock atomic publish refreshes a shortened counter TTL', async () => {
+      const { RedisEventTransport } = await import('../implementations/RedisEventTransport');
+
+      const mockPublisher = createMockPublisher();
+      const mockSubscriber = {
+        on: jest.fn(),
+        subscribe: jest.fn().mockResolvedValue(undefined),
+        unsubscribe: jest.fn().mockResolvedValue(undefined),
+      };
+      const transport = new RedisEventTransport(
+        mockPublisher as unknown as Redis,
+        mockSubscriber as unknown as Redis,
+      );
+      const streamId = 'mock-sequence-ttl-refresh';
+      const sequenceKey = `stream:{${streamId}}:seq`;
+
+      await transport.emitChunk(streamId, { index: 0 });
+      expect(await mockPublisher.ttl(sequenceKey)).toBe(86_400);
+
+      await mockPublisher.expire(sequenceKey, 2);
+      await transport.emitChunk(streamId, { index: 1 });
+
+      expect(await mockPublisher.ttl(sequenceKey)).toBe(86_400);
+      transport.destroy();
+    });
   });
 
   describe('Sequential Event Ordering', () => {
@@ -954,7 +985,7 @@ describe('RedisEventTransport Integration Tests', () => {
    * causing non-deterministic message counts.
    */
   describe('Cross-Replica Sequence Synchronization (#12575)', () => {
-    test('shared counter is cleaned up on stream cleanup', async () => {
+    test('shared counter survives local stream cleanup', async () => {
       if (!ioredisClient) {
         console.warn('Redis not available, skipping test');
         return;
@@ -977,17 +1008,13 @@ describe('RedisEventTransport Integration Tests', () => {
       const valBefore = await ioredisClient.get(key);
       expect(valBefore).toBe('5');
 
-      // Cleanup the stream
+      // Cleanup only this transport's local subscriber state. The Redis counter is
+      // shared by every replica and by later generations that reuse this stream ID.
       transport.cleanup(streamId);
 
-      // Poll for the fire-and-forget DEL to complete (robust under CI load)
-      const start = Date.now();
-      let valAfter: string | null = 'pending';
-      while (valAfter !== null && Date.now() - start < 2000) {
-        await new Promise((resolve) => setTimeout(resolve, 10));
-        valAfter = await ioredisClient.get(key);
-      }
-      expect(valAfter).toBeNull();
+      const valAfter = await ioredisClient.get(key);
+      expect(valAfter).toBe('5');
+      expect(await ioredisClient.ttl(key)).toBeGreaterThan(0);
 
       transport.destroy();
       subscriber.disconnect();

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

@@ -1994,6 +1994,39 @@ describe('RedisJobStore Integration Tests', () => {
       await store.destroy();
     });
 
+    test('pausing for review extends the event sequence TTL to the approval window', async () => {
+      if (!ioredisClient) {
+        return;
+      }
+
+      const { RedisJobStore } = await import('../implementations/RedisJobStore');
+      const store = new RedisJobStore(ioredisClient);
+      await store.initialize();
+
+      const streamId = `sequence-pause-ttl-${Date.now()}`;
+      const sequenceKey = `stream:{${streamId}}:seq`;
+      await store.createJob(streamId, 'sequence-user', streamId);
+      await ioredisClient.set(sequenceKey, '1', 'EX', 1200);
+
+      const approvalWindowSeconds = 48 * 60 * 60;
+      const pendingAction = {
+        ...buildPendingAction(streamId),
+        expiresAt: Date.now() + approvalWindowSeconds * 1000,
+      };
+      const paused = await store.transitionStatus(streamId, {
+        from: 'running',
+        to: 'requires_action',
+        patch: { pendingAction },
+      });
+      expect(paused).toBe(true);
+
+      const ttl = await ioredisClient.ttl(sequenceKey);
+      expect(ttl).toBeGreaterThan(24 * 60 * 60);
+      expect(ttl).toBeLessThanOrEqual(approvalWindowSeconds + 60);
+
+      await store.destroy();
+    });
+
     test('terminal transitions and deleteJob remove the steers key', async () => {
       if (!ioredisClient) {
         return;

packages/api/src/stream/__tests__/helpers/publisher.ts

@@ -2,6 +2,7 @@ export interface MockPublisher {
   publish: jest.Mock;
   incr: jest.Mock;
   expire: jest.Mock;
+  ttl: jest.Mock;
   get: jest.Mock;
   del: jest.Mock;
   eval: jest.Mock;
@@ -10,6 +11,7 @@ export interface MockPublisher {
 /** Mock publisher with Redis command simulation for atomic sequence counters */
 export function createMockPublisher(): MockPublisher {
   const counters = new Map<string, number>();
+  const ttls = new Map<string, number>();
   const publisher: MockPublisher = {
     publish: jest.fn().mockResolvedValue(1),
     incr: jest.fn().mockImplementation((key: string) => {
@@ -17,7 +19,19 @@ export function createMockPublisher(): MockPublisher {
       counters.set(key, current);
       return Promise.resolve(current);
     }),
-    expire: jest.fn().mockResolvedValue(1),
+    expire: jest.fn().mockImplementation((key: string, ttl: number) => {
+      if (!counters.has(key)) {
+        return Promise.resolve(0);
+      }
+      ttls.set(key, ttl);
+      return Promise.resolve(1);
+    }),
+    ttl: jest.fn().mockImplementation((key: string) => {
+      if (!counters.has(key)) {
+        return Promise.resolve(-2);
+      }
+      return Promise.resolve(ttls.get(key) ?? -1);
+    }),
     get: jest.fn().mockImplementation((key: string) => {
       const val = counters.get(key);
       return Promise.resolve(val != null ? String(val) : null);
@@ -25,6 +39,7 @@ export function createMockPublisher(): MockPublisher {
     del: jest.fn().mockImplementation((...keys: string[]) => {
       for (const key of keys) {
         counters.delete(key);
+        ttls.delete(key);
       }
       return Promise.resolve(keys.length);
     }),
@@ -41,14 +56,21 @@ export function createMockPublisher(): MockPublisher {
       _script: string,
       _numKeys: number,
       seqKey: string,
+      jobKey: string,
       channel: string,
       prefix: string,
       suffix: string,
       ttlSeconds: string,
     ) => {
       const val = (await publisher.incr(seqKey)) as number;
-      if (val === 1) {
-        await publisher.expire(seqKey, Number(ttlSeconds));
+      let ttl = Number(ttlSeconds);
+      const seqTtl = (await publisher.ttl(seqKey)) as number;
+      if (seqTtl < Math.floor(ttl / 2)) {
+        const jobTtl = (await publisher.ttl(jobKey)) as number;
+        if (jobTtl > ttl) {
+          ttl = jobTtl;
+        }
+        await publisher.expire(seqKey, ttl);
       }
       const seq = val - 1;
       await publisher.publish(channel, `${prefix}${seq}${suffix}`);

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

@@ -1,14 +1,15 @@
 import { logger } from '@librechat/data-schemas';
 import type { Redis, Cluster } from 'ioredis';
-import { RedisEventTransport } from '~/stream/implementations/RedisEventTransport';
-import { GenerationJobManagerClass } from '~/stream/GenerationJobManager';
-import { createStreamServices } from '~/stream/createStreamServices';
-import { createMockPublisher } from './helpers/publisher';
 import {
   ioredisClient as staticRedisClient,
   keyvRedisClient as staticKeyvClient,
   keyvRedisClientReady,
 } from '~/cache/redisClients';
+import { RedisEventTransport } from '~/stream/implementations/RedisEventTransport';
+import { InMemoryJobStore } from '~/stream/implementations/InMemoryJobStore';
+import { GenerationJobManagerClass } from '~/stream/GenerationJobManager';
+import { createStreamServices } from '~/stream/createStreamServices';
+import { createMockPublisher } from './helpers/publisher';
 
 logger.silent = true;
 
@@ -25,6 +26,11 @@ logger.silent = true;
  * instead of deleting it. The state is fully cleaned up by cleanup() when the
  * job completes.
  *
+ * A second failure mode reused the conversation-scoped stream after one replica
+ * deleted its shared sequence counter. A subscriber still attached elsewhere kept
+ * the old nextSeq and rejected the new turn's restarted sequence as duplicates.
+ * Local cleanup now preserves the shared counter until its bounded Redis TTL expires.
+ *
  * Run with: USE_REDIS=true npx jest reconnect-reorder-desync
  */
 describe('Reconnect Reorder Buffer Desync (Regression)', () => {
@@ -112,6 +118,144 @@ describe('Reconnect Reorder Buffer Desync (Regression)', () => {
       sub2.unsubscribe();
       transport.destroy();
     });
+
+    test('stale unsubscribe cannot detach a replacement stream state', () => {
+      const mockPublisher = createMockPublisher();
+      const mockSubscriber = {
+        on: jest.fn(),
+        subscribe: jest.fn().mockResolvedValue(undefined),
+        unsubscribe: jest.fn().mockResolvedValue(undefined),
+      };
+      const transport = new RedisEventTransport(
+        mockPublisher as unknown as Redis,
+        mockSubscriber as unknown as Redis,
+      );
+      const streamId = 'stale-unsubscribe-test';
+
+      const staleSubscription = transport.subscribe(streamId, { onChunk: () => {} });
+      transport.cleanup(streamId);
+      mockSubscriber.unsubscribe.mockClear();
+
+      const currentSubscription = transport.subscribe(streamId, { onChunk: () => {} });
+      staleSubscription.unsubscribe();
+
+      expect(transport.getSubscriberCount(streamId)).toBe(1);
+      expect(mockSubscriber.unsubscribe).not.toHaveBeenCalled();
+
+      currentSubscription.unsubscribe();
+      transport.destroy();
+    });
+  });
+
+  describe('First-subscriber replay ordering (Unit)', () => {
+    test('does not deliver a buffered event twice while its Redis sequence assignment settles', async () => {
+      const mockPublisher = createMockPublisher();
+      let resolveSequence!: (sequence: number) => void;
+      let markEvalStarted!: () => void;
+      const evalStarted = new Promise<void>((resolve) => {
+        markEvalStarted = resolve;
+      });
+      mockPublisher.eval.mockImplementationOnce(
+        () =>
+          new Promise<number>((resolve) => {
+            resolveSequence = resolve;
+            markEvalStarted();
+          }),
+      );
+      // Redis has already executed INCR + PUBLISH while the command response is in flight.
+      mockPublisher.get.mockResolvedValueOnce('1');
+
+      const mockSubscriber = {
+        on: jest.fn(),
+        subscribe: jest.fn().mockResolvedValue(undefined),
+        unsubscribe: jest.fn().mockResolvedValue(undefined),
+      };
+      const transport = new RedisEventTransport(
+        mockPublisher as unknown as Redis,
+        mockSubscriber as unknown as Redis,
+      );
+      const manager = new GenerationJobManagerClass();
+      manager.configure({
+        jobStore: new InMemoryJobStore(),
+        eventTransport: transport,
+        isRedis: true,
+        cleanupOnComplete: false,
+      });
+
+      const streamId = 'first-subscriber-publish-race';
+      const event = {
+        event: 'on_message_delta',
+        data: { index: 0 },
+      };
+      await manager.createJob(streamId, 'user-1');
+
+      const emitPromise = manager.emitChunk(streamId, event);
+      await evalStarted;
+
+      const received: unknown[] = [];
+      const subscribePromise = manager.subscribe(streamId, (chunk) => received.push(chunk));
+      for (
+        let attempt = 0;
+        attempt < 10 && transport.getSubscriberCount(streamId) === 0;
+        attempt++
+      ) {
+        await new Promise<void>((resolve) => setImmediate(resolve));
+      }
+      expect(transport.getSubscriberCount(streamId)).toBe(1);
+
+      const messageHandler = mockSubscriber.on.mock.calls.find(
+        (call) => call[0] === 'message',
+      )?.[1] as (channel: string, message: string) => void;
+      messageHandler(
+        `stream:{${streamId}}:events`,
+        JSON.stringify({ type: 'chunk', seq: 0, data: event }),
+      );
+      const deliveredBeforeReplay = received.length;
+
+      resolveSequence(0);
+      await Promise.all([emitPromise, subscribePromise]);
+
+      expect(deliveredBeforeReplay).toBe(0);
+      expect(received).toEqual([event]);
+
+      await manager.destroy();
+    });
+
+    test('a failed Redis sync still releases only events beyond the replay frontier', async () => {
+      const mockPublisher = createMockPublisher();
+      mockPublisher.get.mockRejectedValueOnce(new Error('GET failed'));
+      const mockSubscriber = {
+        on: jest.fn(),
+        subscribe: jest.fn().mockResolvedValue(undefined),
+        unsubscribe: jest.fn().mockResolvedValue(undefined),
+      };
+      const transport = new RedisEventTransport(
+        mockPublisher as unknown as Redis,
+        mockSubscriber as unknown as Redis,
+      );
+      const streamId = 'first-subscriber-sync-failure';
+      const chunks: unknown[] = [];
+      transport.subscribe(
+        streamId,
+        {
+          onChunk: (event) => chunks.push(event),
+        },
+        { deferSequenceDelivery: true },
+      );
+
+      const messageHandler = mockSubscriber.on.mock.calls.find(
+        (call) => call[0] === 'message',
+      )?.[1] as (channel: string, message: string) => void;
+      const channel = `stream:{${streamId}}:events`;
+      messageHandler(channel, JSON.stringify({ type: 'chunk', seq: 0, data: { index: 0 } }));
+      messageHandler(channel, JSON.stringify({ type: 'chunk', seq: 1, data: { index: 1 } }));
+      expect(chunks).toEqual([]);
+
+      await expect(transport.syncReorderBuffer(streamId, 1)).rejects.toThrow('GET failed');
+
+      expect(chunks).toEqual([{ index: 1 }]);
+      transport.destroy();
+    });
   });
 
   describe('Reorder buffer sync on reconnect (Unit)', () => {
@@ -291,6 +435,51 @@ describe('Reconnect Reorder Buffer Desync (Regression)', () => {
   });
 
   describe('syncReorderBuffer race: message arrives during async GET window (Unit)', () => {
+    test('stale sync cannot overwrite a replacement stream state', async () => {
+      const mockPublisher = createMockPublisher();
+      const mockSubscriber = {
+        on: jest.fn(),
+        subscribe: jest.fn().mockResolvedValue(undefined),
+        unsubscribe: jest.fn().mockResolvedValue(undefined),
+      };
+      const transport = new RedisEventTransport(
+        mockPublisher as unknown as Redis,
+        mockSubscriber as unknown as Redis,
+      );
+      const streamId = 'stale-sync-replacement-test';
+      transport.subscribe(streamId, { onChunk: () => {} });
+
+      let resolveOldSync!: (value: string | null) => void;
+      mockPublisher.get.mockImplementationOnce(
+        () =>
+          new Promise<string | null>((resolve) => {
+            resolveOldSync = resolve;
+          }),
+      );
+      const oldSync = transport.syncReorderBuffer(streamId);
+
+      transport.cleanup(streamId);
+      const replacementChunks: unknown[] = [];
+      transport.subscribe(streamId, {
+        onChunk: (event) => replacementChunks.push(event),
+      });
+
+      const messageHandler = mockSubscriber.on.mock.calls.find(
+        (call) => call[0] === 'message',
+      )?.[1] as (channel: string, message: string) => void;
+      const channel = `stream:{${streamId}}:events`;
+      messageHandler(channel, JSON.stringify({ type: 'chunk', seq: 0, data: { index: 0 } }));
+
+      // The old GET completes after cleanup with an obsolete high frontier.
+      // It must not advance the replacement state from nextSeq=1 to nextSeq=50.
+      resolveOldSync('50');
+      await oldSync;
+      messageHandler(channel, JSON.stringify({ type: 'chunk', seq: 1, data: { index: 1 } }));
+
+      expect(replacementChunks).toEqual([{ index: 0 }, { index: 1 }]);
+      transport.destroy();
+    });
+
     test('should not drop a chunk that lands in pending while GET is in-flight', async () => {
       const mockPublisher = createMockPublisher();
       const mockSubscriber = {
@@ -350,7 +539,7 @@ describe('Reconnect Reorder Buffer Desync (Regression)', () => {
       transport.destroy();
     });
 
-    test('same-replica: should not drop a live chunk when INCR advances past earlyReplayCount during GET', async () => {
+    test('same-replica: should not drop a live chunk when INCR advances past the replay frontier during GET', async () => {
       const mockPublisher = createMockPublisher();
       const mockSubscriber = {
         on: jest.fn(),
@@ -390,7 +579,7 @@ describe('Reconnect Reorder Buffer Desync (Regression)', () => {
           }),
       );
 
-      // Call syncReorderBuffer with earlyReplayCount=5 (seqs 0–4 were replayed)
+      // Absolute replay frontier is 5 (seqs 0–4 were replayed).
       const syncPromise = transport.syncReorderBuffer(streamId, 5);
 
       // During GET window: LLM emits seq 5 (INCR → counter=6), subscriber receives it
@@ -401,7 +590,7 @@ describe('Reconnect Reorder Buffer Desync (Regression)', () => {
       resolveGet('6');
       await syncPromise;
 
-      // seq 5 MUST be delivered — it's live (seq 5 >= earlyReplayCount 5), not a duplicate.
+      // seq 5 MUST be delivered — it is at the replay frontier, not below it.
       // With the old boolean pruneStaleEntries, 5 < currentSeq(6) would have pruned it.
       expect(chunks.map((c) => (c as { index: number }).index)).toContain(5);
 
@@ -463,7 +652,7 @@ describe('Reconnect Reorder Buffer Desync (Regression)', () => {
       // Now pub/sub for seq 5 arrives AFTER sync completed
       messageHandler(channel, JSON.stringify({ type: 'chunk', seq: 5, data: { index: 5 } }));
 
-      // seq 5 must be delivered — nextSeq should have been capped at earlyReplayCount (5),
+      // seq 5 must be delivered — nextSeq should have been capped at the replay frontier (5),
       // not advanced to currentSeq (6) which would have dropped it.
       expect(chunks.map((c) => (c as { index: number }).index)).toContain(5);
 
@@ -677,5 +866,168 @@ describe('Reconnect Reorder Buffer Desync (Regression)', () => {
 
       await manager.destroy();
     });
+
+    test('mid-generation buffer replay advances to its absolute Redis sequence', async () => {
+      if (!ioredisClient) {
+        console.warn('Redis not available, skipping test');
+        return;
+      }
+
+      const manager = new GenerationJobManagerClass();
+      manager.configure(
+        createStreamServices({
+          useRedis: true,
+          redisClient: ioredisClient,
+        }),
+      );
+      manager.initialize();
+
+      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));
+
+      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 },
+      });
+
+      const resumedEvents: unknown[] = [];
+      const resumedSubscription = await manager.subscribe(streamId, (event) => {
+        resumedEvents.push(event);
+      });
+      await new Promise((resolve) => setTimeout(resolve, 100));
+
+      await manager.emitChunk(streamId, {
+        event: 'on_message_delta',
+        data: { index: 2 },
+      });
+      // Must arrive before the 500 ms reorder timeout.
+      await new Promise((resolve) => setTimeout(resolve, 200));
+
+      expect(
+        resumedEvents.map((event) => (event as { data: { index: number } }).data.index),
+      ).toEqual([1, 2]);
+
+      resumedSubscription?.unsubscribe();
+      await manager.destroy();
+    });
+
+    /**
+     * A producer replica can tear down its local transport after generation 1 while a
+     * subscriber on another replica is still attached with nextSeq=10. Since stream IDs
+     * are conversation IDs, generation 2 reuses the same Redis ordering namespace. The
+     * shared counter must continue at 10; resetting it to 0 makes the lingering consumer
+     * reject every regenerated chunk as an old duplicate.
+     */
+    test('regenerated turn reaches a lingering cross-replica subscriber after producer cleanup', async () => {
+      if (!ioredisClient) {
+        console.warn('Redis not available, skipping test');
+        return;
+      }
+
+      const producer = new GenerationJobManagerClass();
+      const producerServices = createStreamServices({
+        useRedis: true,
+        redisClient: ioredisClient,
+      });
+      producer.configure(producerServices);
+      producer.initialize();
+
+      const consumer = new GenerationJobManagerClass();
+      const consumerServices = createStreamServices({
+        useRedis: true,
+        redisClient: ioredisClient,
+      });
+      consumer.configure(consumerServices);
+      consumer.initialize();
+
+      const streamId = `xrep-regen-${Date.now()}`;
+      const sequenceKey = `stream:{${streamId}}:seq`;
+      await producer.createJob(streamId, 'user-1');
+
+      const firstGeneration: unknown[] = [];
+      const lingering = await consumer.subscribe(streamId, (event) => {
+        firstGeneration.push(event);
+      });
+      await new Promise((resolve) => setTimeout(resolve, 100));
+
+      for (let index = 0; index < 10; index++) {
+        await producer.emitChunk(streamId, {
+          event: 'on_message_delta',
+          data: { generation: 1, index },
+        });
+      }
+      await new Promise((resolve) => setTimeout(resolve, 200));
+      expect(
+        firstGeneration.filter((event) => JSON.stringify(event).includes('"generation":1')),
+      ).toHaveLength(10);
+      expect(await ioredisClient.get(sequenceKey)).toBe('10');
+
+      await producer.completeJob(streamId);
+      producerServices.eventTransport.cleanup(streamId);
+
+      // Local cleanup must not reset a counter that another replica's subscriber
+      // still uses as its ordering frontier.
+      expect(await ioredisClient.get(sequenceKey)).toBe('10');
+
+      await producer.createJob(streamId, 'user-1');
+
+      // Exercise the normal POST-then-SSE path: generation can emit before its
+      // local subscriber attaches, so this event is replayed from earlyEventBuffer.
+      // Its Redis sequence is 10, not 0, because the conversation counter survived.
+      await producer.emitChunk(streamId, {
+        event: 'on_message_delta',
+        data: { generation: 2, index: 0 },
+      });
+
+      const regenerated: unknown[] = [];
+      const secondSubscription = await producer.subscribe(streamId, (event) => {
+        regenerated.push(event);
+      });
+      await new Promise((resolve) => setTimeout(resolve, 100));
+
+      for (let index = 1; index < 5; index++) {
+        await producer.emitChunk(streamId, {
+          event: 'on_message_delta',
+          data: { generation: 2, index },
+        });
+      }
+      // Live chunks must not wait for the 500 ms reorder-buffer force flush.
+      await new Promise((resolve) => setTimeout(resolve, 250));
+
+      expect(
+        regenerated
+          .filter((event) => JSON.stringify(event).includes('"generation":2'))
+          .map((event) => (event as { data: { index: number } }).data.index),
+      ).toEqual([0, 1, 2, 3, 4]);
+      expect(
+        firstGeneration
+          .filter((event) => JSON.stringify(event).includes('"generation":2'))
+          .map((event) => (event as { data: { index: number } }).data.index),
+      ).toEqual([0, 1, 2, 3, 4]);
+      expect(await ioredisClient.get(sequenceKey)).toBe('15');
+
+      lingering?.unsubscribe();
+      secondSubscription?.unsubscribe();
+      await producer.destroy();
+      await consumer.destroy();
+    });
   });
 });

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

@@ -255,4 +255,59 @@ describe('GenerationJobManager - generation abort on reaping', () => {
       jest.useRealTimers();
     }
   });
+
+  it('does not reap a replacement runtime from a stale hasJob result', async () => {
+    const { GenerationJobManagerClass } = await import('../GenerationJobManager');
+    const { InMemoryJobStore } = await import('../implementations/InMemoryJobStore');
+    const { InMemoryEventTransport } = await import('../implementations/InMemoryEventTransport');
+
+    const store = new InMemoryJobStore({ ttlAfterComplete: 60000 });
+    const transport = new InMemoryEventTransport();
+    const manager = new GenerationJobManagerClass();
+    manager.configure({
+      jobStore: store,
+      eventTransport: transport,
+      isRedis: false,
+    });
+    manager.initialize();
+
+    const streamId = 'replacement-during-cleanup';
+    const original = await manager.createJob(streamId, 'user-1', streamId);
+
+    let releaseHasJob!: (exists: boolean) => void;
+    let markHasJobStarted!: () => void;
+    const hasJobStarted = new Promise<void>((resolve) => {
+      markHasJobStarted = resolve;
+    });
+    const hasJobSpy = jest.spyOn(store, 'hasJob').mockImplementationOnce(() => {
+      markHasJobStarted();
+      return new Promise<boolean>((resolve) => {
+        releaseHasJob = resolve;
+      });
+    });
+
+    try {
+      const cleanupPromise = (
+        manager as unknown as {
+          cleanup: () => Promise<void>;
+        }
+      ).cleanup();
+      await hasJobStarted;
+
+      // The old lookup is still in flight while a fresh generation replaces
+      // both the store record and the manager's runtime under the same ID.
+      const replacement = await manager.createJob(streamId, 'user-1', streamId);
+      releaseHasJob(false);
+      await cleanupPromise;
+
+      expect(original.abortController.signal.aborted).toBe(true);
+      expect(replacement.abortController.signal.aborted).toBe(false);
+      expect(await manager.hasJob(streamId)).toBe(true);
+      expect(manager.getRuntimeStats().runtimeStateSize).toBe(1);
+      expect(manager.getRuntimeStats().eventTransportStreams).toBe(1);
+    } finally {
+      hasJobSpy.mockRestore();
+      await manager.destroy();
+    }
+  });
 });

packages/api/src/stream/implementations/RedisEventTransport.ts

@@ -17,6 +17,8 @@ const CHANNELS = {
 const KEYS = {
   /** Atomic sequence counter: shared across all replicas for a given stream */
   sequence: (streamId: string) => `stream:{${streamId}}:seq`,
+  /** Job metadata, used to keep the sequence counter alive for the full job lifetime */
+  job: (streamId: string) => `stream:{${streamId}}:job`,
 };
 
 /**
@@ -48,6 +50,8 @@ interface ReorderBuffer {
   pending: Map<number, PubSubMessage>;
   /** Timeout handle for flushing stale messages */
   flushTimeout: ReturnType<typeof setTimeout> | null;
+  /** Hold sequenced delivery until first-subscriber replay establishes its frontier. */
+  deliveryDeferred: boolean;
 }
 
 /**
@@ -57,21 +61,29 @@ interface ReorderBuffer {
  * re-encoding arbitrary event data would coerce empty arrays to objects and alter float
  * precision. The caller pre-serializes everything around the seq, so this only concatenates.
  *
- * The TTL is set once on the first INCR and never refreshed, so an active stream cannot have
- * its counter reset mid-generation.
+ * The sequence TTL is extend-only. Once it falls below half the safety window, it is refreshed
+ * to the longer of that window and the live job TTL. Checking the job TTL only at that threshold
+ * keeps it off the per-delta hot path. Because stream IDs are conversation IDs, keeping this
+ * counter monotonic across normal cleanup lets a lingering subscriber order later turns.
  *
  * The channel is passed as ARGV, not KEYS: ioredis applies `keyPrefix` to EVAL keys but never
  * to a pub/sub channel, so keying it here would publish to a prefixed channel that no
  * subscriber listens on. PUBLISH is broadcast cluster-wide rather than slot-routed, so it does
  * not need to be a key for Cluster correctness.
  *
- *   KEYS: [sequence]
+ *   KEYS: [sequence, job]
  *   ARGV: [channel, payloadPrefix, payloadSuffix, sequenceTtlSeconds]
  *   RETURNS: the 0-indexed seq assigned to this event
  */
 const PUBLISH_SEQ_LUA =
   'local val = redis.call("INCR", KEYS[1]) ' +
-  'if val == 1 then redis.call("EXPIRE", KEYS[1], tonumber(ARGV[4])) end ' +
+  'local ttl = tonumber(ARGV[4]) ' +
+  'local seqTtl = redis.call("TTL", KEYS[1]) ' +
+  'if seqTtl < math.floor(ttl / 2) then ' +
+  'local jobTtl = redis.call("TTL", KEYS[2]) ' +
+  'if jobTtl > ttl then ttl = jobTtl end ' +
+  'redis.call("EXPIRE", KEYS[1], ttl) ' +
+  'end ' +
   'local seq = val - 1 ' +
   'redis.call("PUBLISH", ARGV[1], ARGV[2] .. string.format("%d", seq) .. ARGV[3]) ' +
   'return seq';
@@ -148,7 +160,7 @@ export class RedisEventTransport implements IEventTransport {
     });
   }
 
-  /** Safety-net TTL (seconds) set once on first INCR. Not refreshed — prevents mid-stream resets. */
+  /** Minimum safety-net TTL in seconds; publishing and pause transitions may only extend it. */
   private static readonly SEQUENCE_TTL_SECONDS = 86400;
 
   /**
@@ -168,8 +180,9 @@ export class RedisEventTransport implements IEventTransport {
   /**
    * Allocate a sequence number and publish, in one Redis round trip.
    *
-   * Keys are deleted explicitly by cleanup() on normal stream teardown; the TTL is a safety
-   * net for orphaned keys from crashed processes.
+   * The shared counter survives local cleanup and expires after its sliding TTL once no
+   * generation is publishing. This bounds storage without resetting another replica's
+   * subscriber frontier between turns.
    */
   private async publishWithSequence(
     streamId: string,
@@ -178,8 +191,9 @@ export class RedisEventTransport implements IEventTransport {
     const [prefix, suffix] = RedisEventTransport.buildPayloadParts(message);
     const seq = await this.publisher.eval(
       PUBLISH_SEQ_LUA,
-      1,
+      2,
       KEYS.sequence(streamId),
+      KEYS.job(streamId),
       CHANNELS.events(streamId),
       prefix,
       suffix,
@@ -198,66 +212,101 @@ export class RedisEventTransport implements IEventTransport {
       }
       state.reorderBuffer.nextSeq = 0;
       state.reorderBuffer.pending.clear();
+      state.reorderBuffer.deliveryDeferred = false;
     }
   }
 
   /**
    * Advance subscriber reorder buffer to the authoritative Redis sequence counter (cross-replica safe).
    *
-   * @param earlyReplayCount - Number of events replayed from earlyEventBuffer (same-replica).
-   *   Pending entries with seq < earlyReplayCount are duplicates and are pruned; entries at or
-   *   above are live chunks from ongoing generation that arrived during the async GET window.
-   *   Using the replay count (not the Redis counter) as the prune cutoff is critical: INCR can
-   *   advance the counter past a live chunk's seq during the GET window, so currentSeq is not
-   *   a safe proxy for "already delivered via earlyEventBuffer."
-   *   When 0/undefined (cross-replica), all pending entries are treated as live and preserved.
+   * @param replayedNextSeq - Absolute Redis sequence immediately after the last event replayed
+   *   from earlyEventBuffer. Pending entries below it were already delivered; entries at or
+   *   above it are live chunks from the ongoing generation. Using the exact replay frontier
+   *   (not the Redis counter) is critical: INCR can advance the counter past a live chunk's
+   *   sequence during the GET window. Undefined means no local replay, so currentSeq is trusted.
    */
-  async syncReorderBuffer(streamId: string, earlyReplayCount = 0): Promise<void> {
-    const key = KEYS.sequence(streamId);
-    const rawStr = await this.publisher.get(key);
-    const parsed = rawStr != null ? parseInt(rawStr, 10) : 0;
-    const currentSeq = Number.isNaN(parsed) ? 0 : parsed;
-    const state = this.streams.get(streamId);
-    if (state) {
-      if (state.reorderBuffer.flushTimeout) {
-        clearTimeout(state.reorderBuffer.flushTimeout);
-        state.reorderBuffer.flushTimeout = null;
+  async syncReorderBuffer(streamId: string, replayedNextSeq?: number): Promise<void> {
+    const initialState = this.streams.get(streamId);
+    try {
+      const key = KEYS.sequence(streamId);
+      const rawStr = await this.publisher.get(key);
+      const parsed = rawStr != null ? parseInt(rawStr, 10) : 0;
+      const currentSeq = Number.isNaN(parsed) ? 0 : parsed;
+      const state = this.streams.get(streamId);
+      // cleanup() may replace this stream's local state while the Redis GET is in
+      // flight. An obsolete snapshot must never move the replacement's frontier.
+      if (state !== initialState) {
+        return;
       }
-      // Prune true duplicates: entries with seq < earlyReplayCount were already delivered
-      // via earlyEventBuffer. Entries at or above are live (possibly from ongoing generation).
-      if (earlyReplayCount > 0) {
-        for (const seq of state.reorderBuffer.pending.keys()) {
-          if (seq < earlyReplayCount) {
-            state.reorderBuffer.pending.delete(seq);
+      if (!state) {
+        return;
+      }
+
+      const buffer = state.reorderBuffer;
+      if (buffer.flushTimeout) {
+        clearTimeout(buffer.flushTimeout);
+        buffer.flushTimeout = null;
+      }
+
+      // Prune true duplicates already delivered via earlyEventBuffer. Entries at or above
+      // the absolute replay frontier are live (possibly from an ongoing generation).
+      if (replayedNextSeq != null) {
+        for (const seq of buffer.pending.keys()) {
+          if (seq < replayedNextSeq) {
+            buffer.pending.delete(seq);
           }
         }
       }
+
       // Set nextSeq from remaining state. Never regress — handleOrderedChunk may have
       // already advanced it during the async GET window.
-      if (state.reorderBuffer.pending.size === 0) {
-        // Same-replica: INCR precedes PUBLISH, so currentSeq may reflect allocated-but-
-        // not-yet-delivered events. Cap at earlyReplayCount to avoid skipping in-flight chunks.
-        // Cross-replica (earlyReplayCount=0): trust the Redis counter.
-        const ceiling = earlyReplayCount > 0 ? earlyReplayCount : currentSeq;
-        state.reorderBuffer.nextSeq = Math.max(state.reorderBuffer.nextSeq, ceiling);
+      if (buffer.pending.size === 0) {
+        // Same-replica replay: INCR precedes PUBLISH, so currentSeq may reflect
+        // allocated-but-not-yet-delivered events. Cap at the exact replay frontier to
+        // avoid skipping in-flight chunks. With no local replay, trust the Redis counter.
+        const ceiling = replayedNextSeq ?? currentSeq;
+        buffer.nextSeq = Math.max(buffer.nextSeq, ceiling);
       } else {
         let minPending = Infinity;
-        for (const seq of state.reorderBuffer.pending.keys()) {
+        for (const seq of buffer.pending.keys()) {
           if (seq < minPending) {
             minPending = seq;
           }
         }
-        state.reorderBuffer.nextSeq = Math.max(
-          state.reorderBuffer.nextSeq,
-          Math.min(currentSeq, minPending),
-        );
-        this.flushPendingMessages(streamId, state);
+        buffer.nextSeq = Math.max(buffer.nextSeq, Math.min(currentSeq, minPending));
       }
+
+      buffer.deliveryDeferred = false;
+      this.flushPendingMessages(streamId, state);
+
       // Re-arm flush timeout if gaps remain after sync — without this,
       // buffered messages could sit indefinitely if no new messages arrive.
-      if (state.reorderBuffer.pending.size > 0) {
+      if (buffer.pending.size > 0) {
         this.scheduleFlushTimeout(streamId, state);
       }
+    } catch (err) {
+      const state = this.streams.get(streamId);
+      // A failed Redis GET must not leave a live subscription permanently paused.
+      // Fall back to normal reorder/timeout behavior and let the caller log the sync error.
+      if (state === initialState && state?.reorderBuffer.deliveryDeferred) {
+        const buffer = state.reorderBuffer;
+        // The local replay frontier remains authoritative even when the shared counter
+        // cannot be read. Drop its pub/sub copies before releasing any later live events.
+        if (replayedNextSeq != null) {
+          for (const seq of buffer.pending.keys()) {
+            if (seq < replayedNextSeq) {
+              buffer.pending.delete(seq);
+            }
+          }
+          buffer.nextSeq = Math.max(buffer.nextSeq, replayedNextSeq);
+        }
+        buffer.deliveryDeferred = false;
+        this.flushPendingMessages(streamId, state);
+        if (buffer.pending.size > 0) {
+          this.scheduleFlushTimeout(streamId, state);
+        }
+      }
+      throw err;
     }
   }
 
@@ -278,7 +327,6 @@ export class RedisEventTransport implements IEventTransport {
 
     try {
       const parsed = JSON.parse(message) as PubSubMessage;
-
       if (parsed.type === EventTypes.CHUNK && parsed.seq != null) {
         this.handleOrderedChunk(streamId, streamState, parsed);
       } else if (
@@ -306,6 +354,11 @@ export class RedisEventTransport implements IEventTransport {
     const buffer = streamState.reorderBuffer;
     const seq = message.seq!;
 
+    if (buffer.deliveryDeferred) {
+      buffer.pending.set(seq, message);
+      return;
+    }
+
     if (seq < buffer.nextSeq) {
       logger.debug(
         `[RedisEventTransport] Dropping duplicate terminal event for stream ${streamId}: seq=${seq}, expected=${buffer.nextSeq}`,
@@ -335,6 +388,11 @@ export class RedisEventTransport implements IEventTransport {
     const buffer = streamState.reorderBuffer;
     const seq = message.seq!;
 
+    if (buffer.deliveryDeferred) {
+      buffer.pending.set(seq, message);
+      return;
+    }
+
     if (seq === buffer.nextSeq) {
       this.deliverMessage(streamState, message);
       buffer.nextSeq++;
@@ -465,6 +523,7 @@ export class RedisEventTransport implements IEventTransport {
       onDone?: (event: unknown) => void;
       onError?: (error: string) => void;
     },
+    options?: { deferSequenceDelivery?: boolean },
   ): { unsubscribe: () => void; ready?: Promise<void> } {
     const channel = CHANNELS.events(streamId);
     const subscriberId = `sub_${++this.subscriberIdCounter}`;
@@ -480,6 +539,7 @@ export class RedisEventTransport implements IEventTransport {
           nextSeq: 0,
           pending: new Map(),
           flushTimeout: null,
+          deliveryDeferred: false,
         },
       });
     }
@@ -490,6 +550,7 @@ export class RedisEventTransport implements IEventTransport {
     // attachment and must not inherit that prior generation's expected seq.
     if (streamState.count === 0) {
       this.resetReorderBuffer(streamId);
+      streamState.reorderBuffer.deliveryDeferred = options?.deferSequenceDelivery === true;
     }
     streamState.count++;
     streamState.handlers.set(subscriberId, handlers);
@@ -512,21 +573,25 @@ export class RedisEventTransport implements IEventTransport {
     return {
       ready: readyPromise,
       unsubscribe: () => {
-        const state = this.streams.get(streamId);
-        if (!state) {
+        // An unsubscribe closure belongs to the exact state and handler created
+        // above. After cleanup + stream reuse, it must not decrement or detach
+        // the replacement subscription that happens to share the same stream ID.
+        if (
+          this.streams.get(streamId) !== streamState ||
+          !streamState.handlers.delete(subscriberId)
+        ) {
           return;
         }
 
-        state.handlers.delete(subscriberId);
-        state.count--;
+        streamState.count--;
 
         // If last subscriber left, unsubscribe from Redis and notify
-        if (state.count === 0) {
+        if (streamState.count === 0) {
           /**
            * Preserve callbacks for reconnect, but drop ordering state from the
            * previous attachment. Reconnects always call syncReorderBuffer(), so
-           * keeping nextSeq here only risks poisoning a later generation when
-           * the shared Redis sequence key has already been reset elsewhere.
+           * keeping a detached subscriber's pending gaps or frontier here can
+           * only delay the next attachment before that authoritative sync.
            */
           this.resetReorderBuffer(streamId);
 
@@ -536,7 +601,7 @@ export class RedisEventTransport implements IEventTransport {
           this.channelSubscriptions.delete(channel);
 
           // Call all-subscribers-left callbacks
-          for (const callback of state.allSubscribersLeftCallbacks) {
+          for (const callback of streamState.allSubscribersLeftCallbacks) {
             try {
               callback();
             } catch (err) {
@@ -564,11 +629,12 @@ export class RedisEventTransport implements IEventTransport {
    * Performance: sequence allocation and publish share one round trip. This runs per streamed
    * delta, so the saved round trip is multiplied by the token count of every response.
    */
-  async emitChunk(streamId: string, event: unknown): Promise<void> {
+  async emitChunk(streamId: string, event: unknown): Promise<number | undefined> {
     try {
-      await this.publishWithSequence(streamId, { type: EventTypes.CHUNK, data: event });
+      return await this.publishWithSequence(streamId, { type: EventTypes.CHUNK, data: event });
     } catch (err) {
       logger.error(`[RedisEventTransport] Failed to publish chunk:`, err);
+      return undefined;
     }
   }
 
@@ -633,6 +699,7 @@ export class RedisEventTransport implements IEventTransport {
           nextSeq: 0,
           pending: new Map(),
           flushTimeout: null,
+          deliveryDeferred: false,
         },
       });
     }
@@ -673,6 +740,7 @@ export class RedisEventTransport implements IEventTransport {
           nextSeq: 0,
           pending: new Map(),
           flushTimeout: null,
+          deliveryDeferred: false,
         },
       };
       this.streams.set(streamId, state);
@@ -700,7 +768,14 @@ export class RedisEventTransport implements IEventTransport {
   }
 
   /**
-   * Cleanup resources for a specific stream.
+   * Cleanup local resources for a specific stream.
+   *
+   * The sequence counter is deliberately left in Redis. A stream ID is currently the
+   * conversation ID, so later turns reuse the same ordering namespace. Another replica
+   * may also still have a subscriber whose reorder buffer is positioned at this counter.
+   * Deleting it here would restart the next producer at zero and make that subscriber
+   * discard the entire next turn as duplicate traffic. The counter's sliding TTL bounds
+   * orphan lifetime.
    */
   cleanup(streamId: string): void {
     const channel = CHANNELS.events(streamId);
@@ -714,13 +789,6 @@ export class RedisEventTransport implements IEventTransport {
 
     this.resetReorderBuffer(streamId);
 
-    // Delete the shared sequence key — safe because cleanup() is only called
-    // when the stream's job is complete (no more publishes will happen).
-    const seqKey = KEYS.sequence(streamId);
-    this.publisher.del(seqKey).catch((err) => {
-      logger.error(`[RedisEventTransport] Failed to delete sequence key ${seqKey}:`, err);
-    });
-
     if (this.channelSubscriptions.has(channel)) {
       this.subscriber.unsubscribe(channel).catch((err) => {
         logger.error(`[RedisEventTransport] Failed to cleanup ${channel}:`, err);
@@ -738,8 +806,7 @@ export class RedisEventTransport implements IEventTransport {
     // Clear all flush timeouts and buffered messages.
     // Sequence keys are NOT deleted here — they are shared across replicas.
     // A shutting-down replica must not nuke the counter for active publishers.
-    // cleanup() deletes keys on normal teardown; a 24h safety-net TTL (set once
-    // at first INCR, never refreshed) caps orphan lifetime on abnormal shutdown.
+    // A sliding 24h safety-net TTL caps orphan lifetime after the last publish.
     for (const [, state] of this.streams) {
       if (state.reorderBuffer.flushTimeout) {
         clearTimeout(state.reorderBuffer.flushTimeout);

packages/api/src/stream/implementations/RedisJobStore.ts

@@ -23,16 +23,16 @@ import { toPendingSteer } from '~/stream/SteeringLifecycle';
 
 /**
  * Atomic compare-and-set on the job hash — the single-winner decision for a
- * status transition. Touches ONLY the job key, which lives on one hash slot, so
- * it is atomic on both single-node and Redis Cluster (cross-slot membership
- * sets are reconciled by the caller AFTER this decides the winner).
+ * status transition. The job and event-sequence keys share the stream hash tag,
+ * so updating the job and extending the counter TTL is atomic on both single-node
+ * Redis and Redis Cluster (cross-slot membership sets are reconciled afterward).
  *
  * Guards on the current `status` and, when ARGV[2] is non-empty, on the flat
  * `pendingActionId` field — so a stale decision targeting a different action
  * loses. On success: removes `clear` fields, writes `status`+patch pairs,
  * refreshes the job-hash TTL. Returns 1 if it fired, 0 otherwise.
  *
- *   KEYS: [job]
+ *   KEYS: [job, eventSequence]
  *   ARGV: [from, expectActionId | "", ttl, hdelCount, ...hdelFields, ...hsetPairs]
  */
 const JOB_CAS_LUA =
@@ -46,6 +46,8 @@ const JOB_CAS_LUA =
   'for i = idx, #ARGV do hset[#hset + 1] = ARGV[i] end ' +
   'if #hset > 0 then redis.call("HSET", KEYS[1], unpack(hset)) end ' +
   'redis.call("EXPIRE", KEYS[1], ttl) ' +
+  'local seqTtl = redis.call("TTL", KEYS[2]) ' +
+  'if seqTtl >= 0 and seqTtl < ttl then redis.call("EXPIRE", KEYS[2], ttl) end ' +
   'return 1';
 
 /**
@@ -278,6 +280,8 @@ const PARKED_RECOVERY_TTL_S: number = 300;
 const KEYS = {
   /** Job metadata: stream:{streamId}:job */
   job: (streamId: string) => `stream:{${streamId}}:job`,
+  /** Pub/sub event sequence counter: stream:{streamId}:seq */
+  sequence: (streamId: string) => `stream:{${streamId}}:seq`,
   /** Chunk stream (Redis Streams): stream:{streamId}:chunks */
   chunks: (streamId: string) => `stream:{${streamId}}:chunks`,
   /** Run steps: stream:{streamId}:runsteps */
@@ -657,8 +661,9 @@ export class RedisJobStore implements IJobStore {
     //    resolves can never both win (and drive the run twice).
     const won = await this.redis.eval(
       JOB_CAS_LUA,
-      1,
+      2,
       key,
+      KEYS.sequence(streamId),
       from,
       expectActionId ?? '',
       String(ttl),

packages/api/src/stream/interfaces/IJobStore.ts

@@ -616,7 +616,13 @@ export interface IJobStore {
  * Implementations can use EventEmitter, Redis Pub/Sub, etc.
  */
 export interface IEventTransport {
-  /** Subscribe to events for a stream. `ready` resolves once the transport can receive messages. */
+  /**
+   * Subscribe to events for a stream. `ready` resolves once the transport can receive messages.
+   *
+   * Redis callers can defer sequenced delivery until `syncReorderBuffer()` establishes the
+   * replay frontier. This prevents pub/sub copies of locally buffered events from racing ahead
+   * of, and then being duplicated by, first-subscriber replay.
+   */
   subscribe(
     streamId: string,
     handlers: {
@@ -624,10 +630,15 @@ export interface IEventTransport {
       onDone?: (event: unknown) => void;
       onError?: (error: string) => void;
     },
+    options?: { deferSequenceDelivery?: boolean },
   ): { unsubscribe: () => void; ready?: Promise<void> };
 
-  /** Publish a chunk event - returns Promise in Redis mode for ordered delivery */
-  emitChunk(streamId: string, event: unknown): void | Promise<void>;
+  /**
+   * Publish a chunk event.
+   * Redis returns the assigned absolute sequence so locally replayed events can
+   * advance a subscriber to the exact ordering frontier.
+   */
+  emitChunk(streamId: string, event: unknown): void | Promise<void | number>;
 
   /** Publish a done event - returns Promise in Redis mode for ordered delivery */
   emitDone(streamId: string, event: unknown): void | Promise<void>;
@@ -661,12 +672,11 @@ export interface IEventTransport {
 
   /**
    * Advance subscriber reorder buffer to match publisher sequence (cross-replica safe).
-   * @param earlyReplayCount - Number of events replayed from earlyEventBuffer (same-replica).
-   *   Pending entries with seq < earlyReplayCount are duplicates and are pruned; entries at or
-   *   above are live chunks that arrived during the async GET window and are preserved.
-   *   When 0/undefined (cross-replica), all pending entries are treated as live.
+   * @param replayedNextSeq - Absolute Redis sequence immediately after the last event replayed
+   *   from the local early-event buffer. Pending entries below it are duplicates; entries at
+   *   or above it are live. Undefined means no local replay, so the Redis counter is trusted.
    */
-  syncReorderBuffer?(streamId: string, earlyReplayCount?: number): void | Promise<void>;
+  syncReorderBuffer?(streamId: string, replayedNextSeq?: number): void | Promise<void>;
 
   /** Cleanup transport resources for a specific stream */
   cleanup(streamId: string): void;