🌊 fix: Prevent Buffered Event Duplication on SSE Resume Connections#12225

danny-avila · 2026-03-14T03:30:22Z

Summary

Fixed a content duplication bug on resumable SSE connections where navigating away from and back to an in-progress agent conversation could cause buffered events to be delivered twice.

Identified the root cause: when a subscriber disconnects, events accumulate in earlyEventBuffer; on reconnect, the route sends a sync event whose aggregatedContent already includes all buffered content, but subscribe() then replays the same buffer on top of it.
Added a syncWasSent local boolean in the /chat/stream/:streamId route handler, set only after res.write() succeeds — skipBufferReplay: true is now passed to subscribe() conditionally on that flag rather than unconditionally on isResume.
Wired GenerationJobManager.markSyncSent() into the route after a successful sync write, unifying the new call-site flag with the existing cross-replica syncSent Redis persistence mechanism that was previously never called.
Added a skipBufferReplay option to GenerationJobManager.subscribe(), which skips replaying earlyEventBuffer while still clearing it to prevent memory accumulation.
Extracted SubscribeOptions as a named, exported interface in packages/api/src/types/stream.ts and updated the subscribe() signature to use t.SubscribeOptions.
Corrected a misleading debug log that stated "(sync already sent)" as fact when that was not guaranteed.
Added a missing @param options parent entry to the subscribe() JSDoc.
Extracted createInMemoryManager(), createRedisManager(), and setupDisconnectedStream() helpers in the integration test suite to eliminate repeated setup across all new tests.
Added getResumeState invariant assertions before each resume subscription to verify accumulated content is available before the buffer is skipped.
Added a multi-reconnect cycle test covering two sequential disconnect/resume cycles to verify state is correctly reset between rounds.
Replaced Record<string, unknown> cast chains throughout the new tests with typed ServerSentEvent[] arrays and structured inline type assertions.
Renamed the default-replay test from the inaccurate "skipBufferReplay is false" to "replay buffer by default when no options are passed".
Replaced the if (!ioredisClient) { console.warn(...); return; } pattern across the entire integration test file with describeRedis / testRedis wrappers (keyed to process.env.USE_REDIS) so Redis tests show as skipped rather than silently passed in environments without Redis, and added a file-level ESLint directive to permit expect() inside the custom testRedis block.

Change Type

Bug fix (non-breaking change which fixes an issue)

Testing

Integration tests are in packages/api/src/stream/__tests__/GenerationJobManager.stream_integration.spec.ts under the Resume: skipBufferReplay prevents duplication describe block.

To run in-memory tests only:

cd packages/api && npx jest GenerationJobManager.stream_integration

To run all tests including Redis:

cd packages/api && USE_REDIS=true REDIS_URI=redis://127.0.0.1:6379 npx jest GenerationJobManager.stream_integration

Test Configuration

Variable	Value
`USE_REDIS`	`true` to enable Redis tests, omit for in-memory only
`REDIS_URI`	`redis://127.0.0.1:6379` (default)

Manual repro: open an agent conversation, navigate away mid-generation, then navigate back via ?resume=true. Prior to this fix, content generated while the tab was inactive would appear twice in the message. After this fix, the sync event delivers the accumulated content once and buffer replay is suppressed.

Checklist

My code adheres to this project's style guidelines
I have performed a self-review of my own code
I have commented in any complex areas of my code
My changes do not introduce new warnings
I have written tests demonstrating that my changes are effective or that my feature works
Local unit tests pass with my changes

Copilot

Pull request overview

Adds a skipBufferReplay option to GenerationJobManager.subscribe() and wires it into the agents SSE stream route to prevent duplicate delivery of buffered events during “resume” reconnections (where a sync payload already includes the accumulated content).

Changes:

Extend GenerationJobManager.subscribe() with an options.skipBufferReplay flag to conditionally skip replaying earlyEventBuffer while still clearing it.
Update /agents/chat/stream/:streamId SSE route to pass skipBufferReplay: true for resume=true connections.
Add integration tests (in-memory + Redis) covering skip vs replay behavior and ensuring the buffer is cleared when skipping.

Reviewed changes

Copilot reviewed 3 out of 3 changed files in this pull request and generated 1 comment.

File	Description
packages/api/src/stream/tests/GenerationJobManager.stream_integration.spec.ts	Adds integration tests validating `skipBufferReplay` behavior in both in-memory and Redis-backed setups.
packages/api/src/stream/GenerationJobManager.ts	Introduces `subscribe(..., options?)` with conditional early buffer replay skipping and buffer clearing.
api/server/routes/agents/index.js	Passes `{ skipBufferReplay: true }` to `subscribe()` for resume connections.

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

You can also share your feedback on Copilot code review. Take the survey.

- Introduced a new option `skipBufferReplay` in the `subscribe` method of `GenerationJobManagerClass` to prevent duplication of events when resuming a connection. - Updated the logic to conditionally skip replaying buffered events if a sync event has already been sent, enhancing the efficiency of event handling during reconnections. - Added integration tests to verify the correct behavior of the new option, ensuring that no buffered events are replayed when `skipBufferReplay` is true, while still allowing for normal replay behavior when false.

…ciently - Modified the `subscribe` method to utilize a new `skipBufferReplay` option, allowing for the prevention of duplicate events during resume connections. - Enhanced the logic in the `chat/stream` route to conditionally skip replaying buffered events if a sync event has already been sent, improving event handling efficiency. - Updated integration tests to verify the correct behavior of the new option, ensuring that no buffered events are replayed when `skipBufferReplay` is true, while maintaining normal replay behavior when false.

- Updated integration tests to conditionally run based on the USE_REDIS environment variable, allowing for better control over Redis-related tests. - Refactored test descriptions to utilize a dynamic `describeRedis` function, improving clarity and organization of tests related to Redis functionality. - Removed redundant checks for Redis availability within individual tests, streamlining the test logic and enhancing readability.

The sync event's else branch (new response message) was missing resetContentHandler() and syncStepMessage() calls, leaving stale handler state that caused subsequent deltas to build on partial content instead of the synced aggregatedContent.

Replaces separate getResumeState() + subscribe() calls with a single subscribeWithResume() that atomically drains earlyEventBuffer between the resume snapshot and the subscribe. In in-memory mode, drained events are returned as pendingEvents for the client to replay after sync. In Redis mode, pendingEvents is empty since chunks are already persisted. The route handler now uses the atomic method for resume connections and extracted shared SSE write helpers to reduce duplication. The client replays any pendingEvents through the existing step/content handlers after applying aggregatedContent from the sync payload.

… buffer The previous implementation drained the entire earlyEventBuffer into pendingEvents, but pre-snapshot events are already reflected in aggregatedContent. Replaying them re-introduced the duplication bug through a different vector. Now records buffer length before getResumeState() and slices from that index, so only events arriving during the async gap are returned as pendingEvents. Also: - Handle pendingEvents when resumeState is null (replay directly) - Hoist duplicate test helpers to shared scope - Remove redundant writableEnded guard in onDone

… event Add test asserting the resume path (skipBufferReplay: true) does NOT emit a created event on cross-replica subscribe — prevents the duplication fix from PR #12225 from regressing. Add explanatory JSDoc on the cross-replica fallback branch documenting which fields are preserved from trackUserMessage() and why sender/isCreatedByUser are hardcoded.

* fix: emit created event from metadata on cross-replica subscribe In multi-instance Redis deployments, the created event (which triggers sidebar conversation creation) was lost when the SSE subscriber connected to a different instance than the one generating. The event was only in the generating instance's local earlyEventBuffer and the Redis pub/sub message was already gone by the time the subscriber's channel was active. When subscribing cross-replica (empty buffer, Redis mode, userMessage already in job metadata), reconstruct and emit the created event directly from stored metadata. * test: add skipBufferReplay regression guard for cross-replica created event Add test asserting the resume path (skipBufferReplay: true) does NOT emit a created event on cross-replica subscribe — prevents the duplication fix from PR #12225 from regressing. Add explanatory JSDoc on the cross-replica fallback branch documenting which fields are preserved from trackUserMessage() and why sender/isCreatedByUser are hardcoded. * refactor: replace as-unknown-as casts with discriminated ServerSentEvent union Split ServerSentEvent into StreamEvent | CreatedEvent | FinalEvent so event shapes are statically typed. Removes all as-unknown-as casts in GenerationJobManager and test file; narrows with proper union members where properties are accessed. * fix: await trackUserMessage before PUBLISH for structural ordering trackUserMessage was fire-and-forget — the HSET for userMessage could theoretically race with the PUBLISH. Await it so the write commits before the pub/sub fires, guaranteeing any cross-replica getJob() after the pub/sub window always finds userMessage in Redis. No-op for non-created events (early return before any async work). * refactor: type CreatedEvent.message explicitly, fix JSDoc and import Give CreatedEvent.message its full known shape instead of Record<string, unknown>. Update sendEvent JSDoc to reflect the discriminated union. Use barrel import in test file. * refactor: type FinalEvent fields with explicit message and conversation shapes Replace Record<string, unknown> on requestMessage, responseMessage, conversation, and runMessages with FinalMessageFields and a typed conversation shape. Captures the known field set used by all final event constructors (abort handler in GenerationJobManager and normal completion in request.js) while allowing extension via index signature for fields contributed by the full TMessage/TConversation schemas. * refactor: narrow trackUserMessage with discriminated union, disambiguate error fields Use 'created' in event to narrow ServerSentEvent to CreatedEvent, eliminating all Record<string, unknown> casts and manual field assertions. Add JSDoc to the two distinct error fields on FinalMessageFields and FinalEvent to prevent confusion. * fix: update cross-replica test to expect created event from metadata The cross-replica subscribe fallback now correctly emits a created event reconstructed from persisted metadata when userMessage exists in the Redis job hash. Replica B receives 4 events (created + 3 deltas) instead of 3.

…anny-avila#12225) * fix: skipBufferReplay for job resume connections - Introduced a new option `skipBufferReplay` in the `subscribe` method of `GenerationJobManagerClass` to prevent duplication of events when resuming a connection. - Updated the logic to conditionally skip replaying buffered events if a sync event has already been sent, enhancing the efficiency of event handling during reconnections. - Added integration tests to verify the correct behavior of the new option, ensuring that no buffered events are replayed when `skipBufferReplay` is true, while still allowing for normal replay behavior when false. * refactor: Update GenerationJobManager to handle sync events more efficiently - Modified the `subscribe` method to utilize a new `skipBufferReplay` option, allowing for the prevention of duplicate events during resume connections. - Enhanced the logic in the `chat/stream` route to conditionally skip replaying buffered events if a sync event has already been sent, improving event handling efficiency. - Updated integration tests to verify the correct behavior of the new option, ensuring that no buffered events are replayed when `skipBufferReplay` is true, while maintaining normal replay behavior when false. * test: Enhance GenerationJobManager integration tests for Redis mode - Updated integration tests to conditionally run based on the USE_REDIS environment variable, allowing for better control over Redis-related tests. - Refactored test descriptions to utilize a dynamic `describeRedis` function, improving clarity and organization of tests related to Redis functionality. - Removed redundant checks for Redis availability within individual tests, streamlining the test logic and enhancing readability. * fix: sync handler state for new messages on resume The sync event's else branch (new response message) was missing resetContentHandler() and syncStepMessage() calls, leaving stale handler state that caused subsequent deltas to build on partial content instead of the synced aggregatedContent. * feat: atomic subscribeWithResume to close resume event gap Replaces separate getResumeState() + subscribe() calls with a single subscribeWithResume() that atomically drains earlyEventBuffer between the resume snapshot and the subscribe. In in-memory mode, drained events are returned as pendingEvents for the client to replay after sync. In Redis mode, pendingEvents is empty since chunks are already persisted. The route handler now uses the atomic method for resume connections and extracted shared SSE write helpers to reduce duplication. The client replays any pendingEvents through the existing step/content handlers after applying aggregatedContent from the sync payload. * fix: only capture gap events in subscribeWithResume, not pre-snapshot buffer The previous implementation drained the entire earlyEventBuffer into pendingEvents, but pre-snapshot events are already reflected in aggregatedContent. Replaying them re-introduced the duplication bug through a different vector. Now records buffer length before getResumeState() and slices from that index, so only events arriving during the async gap are returned as pendingEvents. Also: - Handle pendingEvents when resumeState is null (replay directly) - Hoist duplicate test helpers to shared scope - Remove redundant writableEnded guard in onDone

* fix: emit created event from metadata on cross-replica subscribe In multi-instance Redis deployments, the created event (which triggers sidebar conversation creation) was lost when the SSE subscriber connected to a different instance than the one generating. The event was only in the generating instance's local earlyEventBuffer and the Redis pub/sub message was already gone by the time the subscriber's channel was active. When subscribing cross-replica (empty buffer, Redis mode, userMessage already in job metadata), reconstruct and emit the created event directly from stored metadata. * test: add skipBufferReplay regression guard for cross-replica created event Add test asserting the resume path (skipBufferReplay: true) does NOT emit a created event on cross-replica subscribe — prevents the duplication fix from PR danny-avila#12225 from regressing. Add explanatory JSDoc on the cross-replica fallback branch documenting which fields are preserved from trackUserMessage() and why sender/isCreatedByUser are hardcoded. * refactor: replace as-unknown-as casts with discriminated ServerSentEvent union Split ServerSentEvent into StreamEvent | CreatedEvent | FinalEvent so event shapes are statically typed. Removes all as-unknown-as casts in GenerationJobManager and test file; narrows with proper union members where properties are accessed. * fix: await trackUserMessage before PUBLISH for structural ordering trackUserMessage was fire-and-forget — the HSET for userMessage could theoretically race with the PUBLISH. Await it so the write commits before the pub/sub fires, guaranteeing any cross-replica getJob() after the pub/sub window always finds userMessage in Redis. No-op for non-created events (early return before any async work). * refactor: type CreatedEvent.message explicitly, fix JSDoc and import Give CreatedEvent.message its full known shape instead of Record<string, unknown>. Update sendEvent JSDoc to reflect the discriminated union. Use barrel import in test file. * refactor: type FinalEvent fields with explicit message and conversation shapes Replace Record<string, unknown> on requestMessage, responseMessage, conversation, and runMessages with FinalMessageFields and a typed conversation shape. Captures the known field set used by all final event constructors (abort handler in GenerationJobManager and normal completion in request.js) while allowing extension via index signature for fields contributed by the full TMessage/TConversation schemas. * refactor: narrow trackUserMessage with discriminated union, disambiguate error fields Use 'created' in event to narrow ServerSentEvent to CreatedEvent, eliminating all Record<string, unknown> casts and manual field assertions. Add JSDoc to the two distinct error fields on FinalMessageFields and FinalEvent to prevent confusion. * fix: update cross-replica test to expect created event from metadata The cross-replica subscribe fallback now correctly emits a created event reconstructed from persisted metadata when userMessage exists in the Redis job hash. Replica B receives 4 events (created + 3 deltas) instead of 3.

…anny-avila#12225) * fix: skipBufferReplay for job resume connections - Introduced a new option `skipBufferReplay` in the `subscribe` method of `GenerationJobManagerClass` to prevent duplication of events when resuming a connection. - Updated the logic to conditionally skip replaying buffered events if a sync event has already been sent, enhancing the efficiency of event handling during reconnections. - Added integration tests to verify the correct behavior of the new option, ensuring that no buffered events are replayed when `skipBufferReplay` is true, while still allowing for normal replay behavior when false. * refactor: Update GenerationJobManager to handle sync events more efficiently - Modified the `subscribe` method to utilize a new `skipBufferReplay` option, allowing for the prevention of duplicate events during resume connections. - Enhanced the logic in the `chat/stream` route to conditionally skip replaying buffered events if a sync event has already been sent, improving event handling efficiency. - Updated integration tests to verify the correct behavior of the new option, ensuring that no buffered events are replayed when `skipBufferReplay` is true, while maintaining normal replay behavior when false. * test: Enhance GenerationJobManager integration tests for Redis mode - Updated integration tests to conditionally run based on the USE_REDIS environment variable, allowing for better control over Redis-related tests. - Refactored test descriptions to utilize a dynamic `describeRedis` function, improving clarity and organization of tests related to Redis functionality. - Removed redundant checks for Redis availability within individual tests, streamlining the test logic and enhancing readability. * fix: sync handler state for new messages on resume The sync event's else branch (new response message) was missing resetContentHandler() and syncStepMessage() calls, leaving stale handler state that caused subsequent deltas to build on partial content instead of the synced aggregatedContent. * feat: atomic subscribeWithResume to close resume event gap Replaces separate getResumeState() + subscribe() calls with a single subscribeWithResume() that atomically drains earlyEventBuffer between the resume snapshot and the subscribe. In in-memory mode, drained events are returned as pendingEvents for the client to replay after sync. In Redis mode, pendingEvents is empty since chunks are already persisted. The route handler now uses the atomic method for resume connections and extracted shared SSE write helpers to reduce duplication. The client replays any pendingEvents through the existing step/content handlers after applying aggregatedContent from the sync payload. * fix: only capture gap events in subscribeWithResume, not pre-snapshot buffer The previous implementation drained the entire earlyEventBuffer into pendingEvents, but pre-snapshot events are already reflected in aggregatedContent. Replaying them re-introduced the duplication bug through a different vector. Now records buffer length before getResumeState() and slices from that index, so only events arriving during the async gap are returned as pendingEvents. Also: - Handle pendingEvents when resumeState is null (replay directly) - Hoist duplicate test helpers to shared scope - Remove redundant writableEnded guard in onDone

* fix: emit created event from metadata on cross-replica subscribe In multi-instance Redis deployments, the created event (which triggers sidebar conversation creation) was lost when the SSE subscriber connected to a different instance than the one generating. The event was only in the generating instance's local earlyEventBuffer and the Redis pub/sub message was already gone by the time the subscriber's channel was active. When subscribing cross-replica (empty buffer, Redis mode, userMessage already in job metadata), reconstruct and emit the created event directly from stored metadata. * test: add skipBufferReplay regression guard for cross-replica created event Add test asserting the resume path (skipBufferReplay: true) does NOT emit a created event on cross-replica subscribe — prevents the duplication fix from PR danny-avila#12225 from regressing. Add explanatory JSDoc on the cross-replica fallback branch documenting which fields are preserved from trackUserMessage() and why sender/isCreatedByUser are hardcoded. * refactor: replace as-unknown-as casts with discriminated ServerSentEvent union Split ServerSentEvent into StreamEvent | CreatedEvent | FinalEvent so event shapes are statically typed. Removes all as-unknown-as casts in GenerationJobManager and test file; narrows with proper union members where properties are accessed. * fix: await trackUserMessage before PUBLISH for structural ordering trackUserMessage was fire-and-forget — the HSET for userMessage could theoretically race with the PUBLISH. Await it so the write commits before the pub/sub fires, guaranteeing any cross-replica getJob() after the pub/sub window always finds userMessage in Redis. No-op for non-created events (early return before any async work). * refactor: type CreatedEvent.message explicitly, fix JSDoc and import Give CreatedEvent.message its full known shape instead of Record<string, unknown>. Update sendEvent JSDoc to reflect the discriminated union. Use barrel import in test file. * refactor: type FinalEvent fields with explicit message and conversation shapes Replace Record<string, unknown> on requestMessage, responseMessage, conversation, and runMessages with FinalMessageFields and a typed conversation shape. Captures the known field set used by all final event constructors (abort handler in GenerationJobManager and normal completion in request.js) while allowing extension via index signature for fields contributed by the full TMessage/TConversation schemas. * refactor: narrow trackUserMessage with discriminated union, disambiguate error fields Use 'created' in event to narrow ServerSentEvent to CreatedEvent, eliminating all Record<string, unknown> casts and manual field assertions. Add JSDoc to the two distinct error fields on FinalMessageFields and FinalEvent to prevent confusion. * fix: update cross-replica test to expect created event from metadata The cross-replica subscribe fallback now correctly emits a created event reconstructed from persisted metadata when userMessage exists in the Redis job hash. Replica B receives 4 events (created + 3 deltas) instead of 3.

Copilot AI review requested due to automatic review settings March 14, 2026 03:30

Copilot started reviewing on behalf of danny-avila March 14, 2026 03:30 View session

Copilot AI reviewed Mar 14, 2026

View reviewed changes

Comment thread api/server/routes/agents/index.js Outdated

danny-avila changed the title ~~fix: skipBufferReplay for job resume connections~~ 🌊 fix: Prevent Buffered Event Duplication on SSE Resume Connections Mar 14, 2026

danny-avila force-pushed the fix/resumable-sse-duplication branch 2 times, most recently from 971ba4e to 9a76a98 Compare March 14, 2026 05:46

danny-avila added 6 commits March 14, 2026 03:42

danny-avila force-pushed the fix/resumable-sse-duplication branch from bded1ee to db52902 Compare March 14, 2026 07:43

danny-avila merged commit 7bc793b into dev Mar 14, 2026
14 checks passed

danny-avila deleted the fix/resumable-sse-duplication branch March 14, 2026 14:54

danny-avila mentioned this pull request Mar 15, 2026

🛰️ fix: Cross-Replica Created Event Delivery #12231

Merged

7 tasks

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

🌊 fix: Prevent Buffered Event Duplication on SSE Resume Connections#12225