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.

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.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)
   }
 }
 

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.ts

@@ -16,6 +16,11 @@ 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.
+   */
+  ready(): Promise<void>
   dispose(): void
 }
 
@@ -29,6 +34,7 @@ class RedisPubSubChannel<T> implements PubSubChannel<T> {
   private sub: Redis
   private handlers = new Set<(event: T) => void>()
   private disposed = false
+  private readonly subscribed: Promise<void>
 
   constructor(
     redisUrl: string,
@@ -56,12 +62,17 @@ 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 {
-        logger.info(`Subscribed to ${config.label} channel`)
-      }
+    // Settles on failure too: nothing retries a failed subscribe, so waiting on it would only
+    // hold back every stream on this channel.
+    this.subscribed = new Promise((resolve) => {
+      this.sub.subscribe(config.channel, (err) => {
+        if (err) {
+          logger.error(`Failed to subscribe to ${config.label} channel:`, err)
+        } else {
+          logger.info(`Subscribed to ${config.label} channel`)
+        }
+        resolve()
+      })
     })
 
     this.sub.on('message', (channel: string, message: string) => {
@@ -95,6 +106,10 @@ class RedisPubSubChannel<T> implements PubSubChannel<T> {
     }
   }
 
+  ready(): Promise<void> {
+    return this.subscribed
+  }
+
   dispose(): void {
     this.disposed = true
     this.handlers.clear()
@@ -130,6 +145,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`)

apps/sim/lib/events/sse-endpoint.test.ts

@@ -2,10 +2,12 @@ import { authMockFns, permissionsMock, permissionsMockFns } from '@sim/testing'
 import { NextRequest } from 'next/server'
 import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
 import {
+  createSSEStream,
   createWorkspaceSSE,
   HEARTBEAT_INTERVAL_MS,
   MAX_CONNECTION_MS,
   MAX_UNDRAINED_CHUNKS,
+  OPENED_COMMENT,
   ROTATION_GRACE_MS,
 } from '@/lib/events/sse-endpoint'
 
@@ -65,6 +67,16 @@ describe('createWorkspaceSSE', () => {
     vi.useRealTimers()
   })
 
+  it('starts the response before the first heartbeat', async () => {
+    const { body } = await openConnection()
+    const chunks: string[] = []
+    void collect(body, chunks)
+
+    await vi.advanceTimersByTimeAsync(HEARTBEAT_INTERVAL_MS - 1)
+
+    expect(chunks).toEqual([OPENED_COMMENT])
+  })
+
   it('announces rotation before releasing the old connection', async () => {
     const { body, unsubscribe } = await openConnection()
     const chunks: string[] = []
@@ -155,3 +167,62 @@ describe('createWorkspaceSSE', () => {
     expect(unsubscribe).toHaveBeenCalledTimes(1)
   })
 })
+
+describe('createSSEStream', () => {
+  beforeEach(() => {
+    vi.useFakeTimers()
+  })
+
+  afterEach(() => {
+    vi.useRealTimers()
+  })
+
+  it('opens only once its subscriptions receive events', async () => {
+    let live: () => void = () => {}
+    const response = createSSEStream(new NextRequest(new URL('https://sim.test/api/test/stream')), {
+      label: 'test',
+      subscriptions: [
+        {
+          subscribe: () => () => {},
+          ready: () =>
+            new Promise<void>((resolve) => {
+              live = resolve
+            }),
+        },
+      ],
+    })
+    const chunks: string[] = []
+    void collect(response.body as ReadableStream<Uint8Array>, chunks)
+    await vi.advanceTimersByTimeAsync(1_000)
+    expect(chunks).toEqual([])
+
+    live()
+    await vi.advanceTimersByTimeAsync(0)
+
+    expect(chunks).toEqual([OPENED_COMMENT])
+  })
+
+  it('delivers a revalidated event as soon as it is published', async () => {
+    let publish: (eventName: string, data: Record<string, unknown>) => void = () => {}
+    const response = createSSEStream(new NextRequest(new URL('https://sim.test/api/test/stream')), {
+      label: 'test',
+      revalidate: async () => {},
+      subscriptions: [
+        {
+          subscribe: (send) => {
+            publish = send
+            return () => {}
+          },
+        },
+      ],
+    })
+    const chunks: string[] = []
+    void collect(response.body as ReadableStream<Uint8Array>, chunks)
+    await vi.advanceTimersByTimeAsync(0)
+
+    publish('inbox_changed', { reason: 'call' })
+    await vi.advanceTimersByTimeAsync(0)
+
+    expect(chunks).toEqual([OPENED_COMMENT, 'event: inbox_changed\ndata: {"reason":"call"}\n\n'])
+  })
+})

apps/sim/lib/events/sse-endpoint.ts

@@ -19,6 +19,8 @@ interface SSESubscription {
     workspaceId: string,
     send: (eventName: string, data: Record<string, unknown>) => void
   ): () => void
+  /** Settles once the subscription receives events; the stream is announced only after it. */
+  ready?: () => Promise<void>
 }
 
 interface WorkspaceSSEConfig {
@@ -30,6 +32,9 @@ const encoder = new TextEncoder()
 
 export const HEARTBEAT_INTERVAL_MS = 30_000
 
+/** Written once a stream's subscriptions are live; clients ignore comments. */
+export const OPENED_COMMENT = ': connected\n\n'
+
 /**
  * Starts a make-before-break rotation for one connection. Healthy clients open
  * a replacement before this stream closes; orphaned streams are released after
@@ -83,6 +88,7 @@ export function createWorkspaceSSE(config: WorkspaceSSEConfig) {
       label: `${config.label}:workspace:${workspaceId}`,
       subscriptions: config.subscriptions.map((subscription) => ({
         subscribe: (send) => subscription.subscribe(workspaceId, send),
+        ready: subscription.ready,
       })),
     })
   }
@@ -92,6 +98,8 @@ interface SSEStreamConfig {
   label: string
   subscriptions: Array<{
     subscribe(send: (eventName: string, data: Record<string, unknown>) => void): () => void
+    /** Settles once the subscription receives events; the stream is announced only after it. */
+    ready?: () => Promise<void>
   }>
   /** Rechecks a long-lived authorization before each publication and on heartbeats. */
   revalidate?: () => Promise<void>
@@ -224,6 +232,13 @@ export function createSSEStream(request: NextRequest, config: SSEStreamConfig):
         })
         teardowns.push(() => listenerScope.abort())
 
+        // The runtime sends the status and headers with the first body chunk, so without this the
+        // response would not start until the first event or heartbeat. A client reads its state
+        // once the stream opens, so it opens only when no later event can be missed.
+        void Promise.all(config.subscriptions.map((subscription) => subscription.ready?.())).then(
+          () => enqueue(OPENED_COMMENT),
+          () => close('subscription_failed')
+        )
         logger.info(`SSE connection opened for ${config.label}`)
       } catch (error) {
         cleanup('setup_failed')

apps/sim/lib/mcp/pubsub.ts

@@ -19,6 +19,8 @@ interface McpPubSubAdapter {
   publishWorkflowToolsChanged(event: WorkflowToolsChangedEvent): void
   onToolsChanged(handler: (event: ToolsChangedEvent) => void): () => void
   onWorkflowToolsChanged(handler: (event: WorkflowToolsChangedEvent) => void): () => void
+  /** Settles once this process receives both channels' events. */
+  ready(): Promise<void>
   dispose(): void
 }
 
@@ -60,6 +62,9 @@ export const mcpPubSub: McpPubSubAdapter | null =
         publishWorkflowToolsChanged: (event) => workflowToolsChannel.publish(event),
         onToolsChanged: (handler) => toolsChannel.subscribe(handler),
         onWorkflowToolsChanged: (handler) => workflowToolsChannel.subscribe(handler),
+        ready: async () => {
+          await Promise.all([toolsChannel.ready(), workflowToolsChannel.ready()])
+        },
         dispose: () => {
           toolsChannel.dispose()
           workflowToolsChannel.dispose()