fix(events): start every SSE response once its subscriptions are live (#8700)

* fix(events): start every SSE response once its subscriptions are live

Next sends a route's status and headers with the first body chunk, and
createSSEStream wrote nothing until an event or the 30 s heartbeat. A client
therefore saw no response for up to 30 s after connecting: the desktop inbox
doorbell E2E waited 32-34 s for its stream to open in every run, and Sim
desktop's doorbell gives up on a response that has not started within 15 s.

- createSSEStream writes a `: connected` comment as soon as every
  subscription is live, which starts the response at once.
- PubSubChannel exposes ready(), settled when the process's Redis SUBSCRIBE
  completes, so a client that reads its state on open cannot miss an event
  published before the channel was listening. The desktop doorbell, chat
  status and MCP streams wait on their channels.
- The desktop inbox E2E asserts the doorbell opens within the desktop's 15 s
  handshake and that a ring arrives within 2 s of the open.

* fix(events): hold events until the stream opens and bound the wait

- Events wait for the opening gate too, so one published while another
  subscription is still connecting cannot start the response early.
- The gate opens at OPEN_DEADLINE_MS (5 s) when a subscription is not live,
  so a Redis outage degrades to an open stream instead of a silent one.
- PubSubChannel subscribes on every ready connection and is not ready again
  after a dropped one until it has resubscribed.
- The desktop inbox E2E compiles the doorbell route with a refused request
  before timing its open.

* fix(events): keep a channel not ready while its subscribe fails or answers a closed connection

A failed SUBSCRIBE no longer settles ready(): the stream's opening deadline
bounds the wait, and the next connection subscribes again. A subscribe
answered after its connection closed is ignored, so it cannot mark a
reconnected channel ready before that connection has subscribed.

* fix(events): write each stream's events in order and release a stream closed before it opens

- Every event goes through one ordered write chain: it is authorized as soon
  as it arrives, so concurrent events still share one check, but written only
  after every event before it. An event queued before the stream opened can
  no longer be overtaken by one that arrived during its authorization.
- Closing settles the opening gate as well as clearing its deadline, so a
  stream closed before it opens is not kept alive by a subscription that
  never becomes ready.
- Tests cover event order, retention after an early close, timer cleanup,
  and the ready wiring of the mothership and MCP event routes.

Author: waleedlatif1

apps/sim/app/api/desktop/inbox/stream/route.ts

@@ -32,7 +32,7 @@ export const GET = withRouteHandler(async (request: NextRequest) => {
     return createSSEStream(request, {
       label: 'desktop-inbox',
       revalidate: inbox.revalidate,
-      subscriptions: [{ subscribe: inbox.subscribe }],
+      subscriptions: [{ subscribe: inbox.subscribe, ready: inbox.ready }],
     })
   } catch (error) {
     if (error instanceof InternalUnauthenticatedError)

apps/sim/app/api/mcp/events/route.test.ts

@@ -0,0 +1,50 @@
+import { authMockFns, permissionsMock, permissionsMockFns } from '@sim/testing'
+import { mcpPubsubMock, mcpPubsubMockFns } from '@sim/testing/mocks/mcp-pubsub.mock'
+import { NextRequest } from 'next/server'
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
+import { OPENED_COMMENT } from '@/lib/events/sse-endpoint'
+
+vi.mock('@/lib/mcp/pubsub', () => mcpPubsubMock)
+vi.mock('@/lib/mcp/connection-manager', () => ({
+  mcpConnectionManager: { subscribe: () => () => {} },
+}))
+vi.mock('@/lib/workspaces/permissions/utils', () => permissionsMock)
+
+import { GET } from '@/app/api/mcp/events/route'
+
+describe('MCP tool-change event stream', () => {
+  beforeEach(() => {
+    vi.useFakeTimers()
+    authMockFns.mockGetSession.mockResolvedValue({ user: { id: 'user-1' } })
+    permissionsMockFns.mockGetUserEntityPermissions.mockResolvedValue('read')
+  })
+
+  afterEach(() => {
+    vi.useRealTimers()
+  })
+
+  it('opens once tool-change events reach this process', async () => {
+    let live: () => void = () => {}
+    const ready = new Promise<void>((resolve) => {
+      live = resolve
+    })
+    mcpPubsubMockFns.mockReady.mockReturnValue(ready)
+    const response = await GET(new NextRequest('http://localhost/api/mcp/events?workspaceId=ws-1'))
+    let first: string | undefined
+    if (!response.body) throw new Error('The event stream has no body')
+    void response.body
+      .getReader()
+      .read()
+      .then(({ value }) => {
+        first = new TextDecoder().decode(value)
+      })
+
+    await vi.advanceTimersByTimeAsync(1_000)
+    expect(first).toBeUndefined()
+    expect(mcpPubsubMockFns.mockReady).toHaveBeenCalledTimes(2)
+
+    live()
+    await vi.advanceTimersByTimeAsync(0)
+    expect(first).toBe(OPENED_COMMENT)
+  })
+})

apps/sim/app/api/mcp/events/route.ts

@@ -32,6 +32,7 @@ const mcpEventsHandler = createWorkspaceSSE({
           })
         })
       },
+      ready: async () => mcpPubSub?.ready(),
     },
     {
       subscribe: (workspaceId, send) => {
@@ -45,6 +46,7 @@ const mcpEventsHandler = createWorkspaceSSE({
           })
         })
       },
+      ready: async () => mcpPubSub?.ready(),
     },
   ],
 })

apps/sim/app/api/mothership/events/route.test.ts

@@ -16,7 +16,7 @@ import {
 import { NextRequest } from 'next/server'
 import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
 import { OrchestrationError } from '@/lib/core/orchestration/types'
-import { HEARTBEAT_INTERVAL_MS } from '@/lib/events/sse-endpoint'
+import { HEARTBEAT_INTERVAL_MS, OPENED_COMMENT } from '@/lib/events/sse-endpoint'
 import type { ChatStatusEvent } from '@/lib/mothership/chat-status'
 import { PermissionGroupCapabilityError } from '@/lib/permission-groups/capability-error'
 
@@ -41,13 +41,15 @@ function emit(event: ChatStatusEvent) {
   handler(event)
 }
 
+/** Every chunk after the stream's opening comment. */
 async function collect(body: ReadableStream<Uint8Array>, chunks: string[]) {
   const reader = body.getReader()
   const decoder = new TextDecoder()
   while (true) {
     const { done, value } = await reader.read()
     if (done) return
-    chunks.push(decoder.decode(value))
+    const chunk = decoder.decode(value)
+    if (chunk !== OPENED_COMMENT) chunks.push(chunk)
   }
 }
 
@@ -154,6 +156,7 @@ describe('Mothership owner-scoped event stream', () => {
     const response = await GET(request('organizationId=org-1'))
     const chunks: string[] = []
     const collected = collect(response.body!, chunks)
+    await vi.advanceTimersByTimeAsync(0)
     let authorizeDone: (() => void) | undefined
     authorize.mockReturnValueOnce(
       new Promise<void>((resolve) => {
@@ -171,11 +174,40 @@ describe('Mothership owner-scoped event stream', () => {
     expect(chunks).toEqual([])
   })
 
+  it.each(['workspaceId=ws-1', 'organizationId=org-1'])(
+    'opens the %s stream once chat status events reach this process',
+    async (query) => {
+      let live: () => void = () => {}
+      mothershipChatStatusMockFns.mockReady.mockReturnValueOnce(
+        new Promise<void>((resolve) => {
+          live = resolve
+        })
+      )
+      const response = await GET(request(query))
+      let first: string | undefined
+      if (!response.body) throw new Error('The event stream has no body')
+      void response.body
+        .getReader()
+        .read()
+        .then(({ value }) => {
+          first = new TextDecoder().decode(value)
+        })
+
+      await vi.advanceTimersByTimeAsync(1_000)
+      expect(first).toBeUndefined()
+
+      live()
+      await vi.advanceTimersByTimeAsync(0)
+      expect(first).toBe(OPENED_COMMENT)
+    }
+  )
+
   it('preserves workspace status events and excludes organization events', async () => {
     const abort = new AbortController()
     const response = await GET(request('workspaceId=ws-1', abort.signal))
     const chunks: string[] = []
     const collected = collect(response.body!, chunks)
+    await vi.advanceTimersByTimeAsync(0)
     emit({ organizationId: 'org-1', userId: 'user-1', chatId: 'org-chat', type: 'created' })
     emit({ workspaceId: 'ws-2', chatId: 'other-workspace-chat', type: 'created' })
     emit({ workspaceId: 'ws-1', chatId: 'workspace-chat', type: 'renamed' })

apps/sim/app/api/mothership/events/route.ts

@@ -42,6 +42,7 @@ const mothershipEventsHandler = createWorkspaceSSE({
           })
         })
       },
+      ready: async () => chatPubSub?.ready(),
     },
   ],
 })
@@ -80,6 +81,7 @@ export const GET = withRouteHandler(async (request: NextRequest) => {
                 timestamp: Date.now(),
               })
             }) ?? (() => {}),
+          ready: async () => chatPubSub?.ready(),
         },
       ],
     })

apps/sim/lib/desktop/application/executor.ts

@@ -15,6 +15,7 @@ import {
 } from '@/lib/desktop/executor/constants'
 import {
   type DesktopInboxChangeReason,
+  desktopInboxDoorbellReady,
   onDesktopInboxDoorbell,
 } from '@/lib/desktop/executor/doorbell'
 import {
@@ -201,6 +202,7 @@ export const openDesktopInboxStream = defineAuthorizedCredentialUserUseCase({
         onDesktopInboxDoorbell(deviceId, (reason: DesktopInboxChangeReason) =>
           send('inbox_changed', { reason })
         ),
+      ready: desktopInboxDoorbellReady,
     }
   },
 })

apps/sim/lib/desktop/executor/doorbell.ts

@@ -43,6 +43,11 @@ export function ringDesktopInbox(deviceId: string, reason: DesktopInboxChangeRea
   }
 }
 
+/** Settles once this process hears every ring. */
+export function desktopInboxDoorbellReady(): Promise<void> {
+  return channel().ready()
+}
+
 /** Subscribes to one device's doorbell; returns the unsubscribe. */
 export function onDesktopInboxDoorbell(
   deviceId: string,

apps/sim/lib/events/pubsub.test.ts

@@ -0,0 +1,97 @@
+import { EventEmitter } from 'node:events'
+import { redisConfigMockFns } from '@sim/testing/mocks/redis-config.mock'
+import { beforeEach, describe, expect, it, vi } from 'vitest'
+
+const { clients } = vi.hoisted(() => ({
+  clients: [] as Array<EventEmitter & { subscribed: Array<(err: Error | null) => void> }>,
+}))
+
+vi.mock('ioredis', () => ({
+  default: class extends EventEmitter {
+    subscribed: Array<(err: Error | null) => void> = []
+
+    constructor() {
+      super()
+      clients.push(this)
+    }
+
+    subscribe(_channel: string, done: (err: Error | null) => void) {
+      this.subscribed.push(done)
+    }
+
+    publish = vi.fn(async () => 1)
+    unsubscribe = vi.fn(async () => undefined)
+    quit = vi.fn(async () => 'OK')
+  },
+}))
+
+import { createPubSubChannel } from '@/lib/events/pubsub'
+
+/** Whether the promise has settled by the time already-queued work runs. */
+async function settled(promise: Promise<void>): Promise<boolean> {
+  return Promise.race([promise.then(() => true), Promise.resolve().then(() => false)])
+}
+
+/** The subscriber connection: the channel opens its publisher first. */
+function subscriber() {
+  const client = clients.at(-1)
+  if (!client) throw new Error('no Redis client')
+  return client
+}
+
+describe('createPubSubChannel over Redis', () => {
+  beforeEach(() => {
+    clients.length = 0
+    redisConfigMockFns.mockGetConfiguredRedisUrl.mockReturnValue('redis://localhost:6379')
+  })
+
+  it('is ready only once its connection has subscribed', async () => {
+    const channel = createPubSubChannel({ channel: 'test', label: 'Test' })
+    expect(await settled(channel.ready())).toBe(false)
+
+    subscriber().emit('ready')
+    expect(await settled(channel.ready())).toBe(false)
+
+    subscriber().subscribed[0](null)
+    expect(await settled(channel.ready())).toBe(true)
+  })
+
+  it('is not ready again until a dropped connection has resubscribed', async () => {
+    const channel = createPubSubChannel({ channel: 'test', label: 'Test' })
+    subscriber().emit('ready')
+    subscriber().subscribed[0](null)
+
+    subscriber().emit('close')
+    expect(await settled(channel.ready())).toBe(false)
+
+    subscriber().emit('ready')
+    subscriber().subscribed[1](null)
+    expect(await settled(channel.ready())).toBe(true)
+  })
+
+  it('is not ready while its subscribe fails, and retries on the next connection', async () => {
+    const channel = createPubSubChannel({ channel: 'test', label: 'Test' })
+    subscriber().emit('ready')
+
+    subscriber().subscribed[0](new Error('NOPERM'))
+    expect(await settled(channel.ready())).toBe(false)
+
+    subscriber().emit('close')
+    subscriber().emit('ready')
+    subscriber().subscribed[1](null)
+    expect(await settled(channel.ready())).toBe(true)
+  })
+
+  it('ignores a subscribe answered after its connection closed', async () => {
+    const channel = createPubSubChannel({ channel: 'test', label: 'Test' })
+    subscriber().emit('ready')
+
+    subscriber().emit('close')
+    subscriber().subscribed[0](null)
+    expect(await settled(channel.ready())).toBe(false)
+
+    subscriber().emit('ready')
+    subscriber().subscribed[1](null)
+    expect(await settled(channel.ready())).toBe(true)
+  })
+})

apps/sim/lib/events/pubsub.ts

@@ -16,6 +16,12 @@ const logger = createLogger('PubSub')
 export interface PubSubChannel<T> {
   publish(event: T): void
   subscribe(handler: (event: T) => void): () => void
+  /**
+   * Settles once this process receives the channel's publications; anything published before
+   * then reaches no subscriber here. Stays pending while the channel cannot subscribe, so a caller
+   * that must not wait indefinitely bounds the wait.
+   */
+  ready(): Promise<void>
   dispose(): void
 }
 
@@ -29,6 +35,12 @@ class RedisPubSubChannel<T> implements PubSubChannel<T> {
   private sub: Redis
   private handlers = new Set<(event: T) => void>()
   private disposed = false
+  /** Whether the current connection has subscribed; a dropped connection has to again. */
+  private listening = false
+  /** Counts closed connections, so a subscribe answered after its connection closed is ignored. */
+  private closedConnections = 0
+  private subscribed: Promise<void> = Promise.resolve()
+  private markSubscribed: () => void = noop
 
   constructor(
     redisUrl: string,
@@ -56,12 +68,28 @@ class RedisPubSubChannel<T> implements PubSubChannel<T> {
     this.pub.on('connect', () => logger.info(`${config.label} publish client connected`))
     this.sub.on('connect', () => logger.info(`${config.label} subscribe client connected`))
 
-    this.sub.subscribe(config.channel, (err) => {
-      if (err) {
-        logger.error(`Failed to subscribe to ${config.label} channel:`, err)
-      } else {
+    this.awaitSubscription()
+    // Subscribes on every ready connection: ioredis resubscribes after a reconnect on its own but
+    // does not report when that lands, and SUBSCRIBE is idempotent. A failed subscribe leaves the
+    // channel not ready; the next connection tries again.
+    this.sub.on('ready', () => {
+      const connection = this.closedConnections
+      this.sub.subscribe(config.channel, (err) => {
+        if (connection !== this.closedConnections) return
+        if (err) {
+          logger.error(`Failed to subscribe to ${config.label} channel:`, err)
+          return
+        }
+        this.listening = true
         logger.info(`Subscribed to ${config.label} channel`)
-      }
+        this.markSubscribed()
+      })
+    })
+    this.sub.on('close', () => {
+      this.closedConnections += 1
+      if (!this.listening) return
+      this.listening = false
+      this.awaitSubscription()
     })
 
     this.sub.on('message', (channel: string, message: string) => {
@@ -95,6 +123,16 @@ class RedisPubSubChannel<T> implements PubSubChannel<T> {
     }
   }
 
+  ready(): Promise<void> {
+    return this.subscribed
+  }
+
+  private awaitSubscription(): void {
+    this.subscribed = new Promise((resolve) => {
+      this.markSubscribed = resolve
+    })
+  }
+
   dispose(): void {
     this.disposed = true
     this.handlers.clear()
@@ -130,6 +168,10 @@ class LocalPubSubChannel<T> implements PubSubChannel<T> {
     }
   }
 
+  ready(): Promise<void> {
+    return Promise.resolve()
+  }
+
   dispose(): void {
     this.emitter.removeAllListeners()
     logger.info(`${this.config.label} local pub/sub disposed`)