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/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/mothership/events/route.test.ts

@@ -174,6 +174,34 @@ 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))

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

@@ -1,4 +1,7 @@
+import { setFlagsFromString } from 'node:v8'
+import { runInNewContext } from 'node:vm'
 import { authMockFns, permissionsMock, permissionsMockFns } from '@sim/testing'
+import { sleep } from '@sim/utils/helpers'
 import { NextRequest } from 'next/server'
 import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
 import {
@@ -248,6 +251,62 @@ describe('createSSEStream', () => {
     expect(chunks).toEqual([OPENED_COMMENT])
   })
 
+  it('writes revalidated events in the order they arrive', async () => {
+    let live: () => void = () => {}
+    let authorized: () => void = () => {}
+    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: () =>
+        new Promise<void>((resolve) => {
+          authorized = resolve
+        }),
+      subscriptions: [
+        {
+          subscribe: (send) => {
+            publish = send
+            return () => {}
+          },
+          ready: () =>
+            new Promise<void>((resolve) => {
+              live = resolve
+            }),
+        },
+      ],
+    })
+    const chunks: string[] = []
+    void collect(response.body as ReadableStream<Uint8Array>, chunks)
+
+    publish('changed', { n: 1 })
+    live()
+    await vi.advanceTimersByTimeAsync(0)
+    publish('changed', { n: 2 })
+    authorized()
+    await vi.advanceTimersByTimeAsync(0)
+
+    expect(chunks).toEqual([
+      OPENED_COMMENT,
+      'event: changed\ndata: {"n":1}\n\n',
+      'event: changed\ndata: {"n":2}\n\n',
+    ])
+  })
+
+  it('clears its timers when it closes before it opens', async () => {
+    const controller = new AbortController()
+    createSSEStream(
+      new NextRequest(new URL('https://sim.test/api/test/stream'), { signal: controller.signal }),
+      {
+        label: 'test',
+        subscriptions: [{ subscribe: () => () => {}, ready: () => new Promise<void>(() => {}) }],
+      }
+    )
+    expect(vi.getTimerCount()).toBe(2)
+
+    controller.abort()
+
+    expect(vi.getTimerCount()).toBe(0)
+  })
+
   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')), {
@@ -272,3 +331,29 @@ describe('createSSEStream', () => {
     expect(chunks).toEqual([OPENED_COMMENT, 'event: inbox_changed\ndata: {"reason":"call"}\n\n'])
   })
 })
+
+describe('createSSEStream retention', () => {
+  /** Opens a stream and closes it before it opens; only a weak reference to it escapes. */
+  function closedBeforeOpening(ready: Promise<void>): WeakRef<object> {
+    const controller = new AbortController()
+    const response = createSSEStream(
+      new NextRequest(new URL('https://sim.test/api/test/stream'), { signal: controller.signal }),
+      { label: 'test', subscriptions: [{ subscribe: () => () => {}, ready: () => ready }] }
+    )
+    controller.abort()
+    return new WeakRef(response.body as object)
+  }
+
+  it('does not stay reachable from a subscription that never becomes ready', async () => {
+    setFlagsFromString('--expose_gc')
+    const collectGarbage = runInNewContext('gc') as () => void
+    const neverReady = new Promise<void>(() => {})
+
+    const stream = closedBeforeOpening(neverReady)
+    await sleep(0)
+    collectGarbage()
+
+    expect(stream.deref()).toBeUndefined()
+    expect(neverReady).toBeInstanceOf(Promise)
+  })
+})

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

@@ -8,6 +8,7 @@
 import type { SessionPrincipal } from '@sim/auth/principal'
 import { createLogger } from '@sim/logger'
 import { getErrorMessage } from '@sim/utils/errors'
+import { noop } from '@sim/utils/helpers'
 import { randomFloat } from '@sim/utils/random'
 import type { NextRequest } from 'next/server'
 import { getSession } from '@/lib/auth'
@@ -164,14 +165,17 @@ export function createSSEStream(request: NextRequest, config: SSEStreamConfig):
         })
         return authorization
       }
-      /** Settles when the stream announces itself; events wait for it so none opens it early. */
-      let opening: Promise<void> = Promise.resolve()
+      /**
+       * The stream's writes in order: the opening, then each event once it is authorized. An event
+       * waits here while the stream has not opened or an earlier event is still being written.
+       */
+      let writes: Promise<void> = Promise.resolve()
       let opened = false
       let pendingEvents = 0
       const send = (eventName: string, data: Record<string, unknown>) => {
         if (cleaned) return
         const payload = `event: ${eventName}\ndata: ${JSON.stringify(data)}\n\n`
-        if (opened && !config.revalidate) {
+        if (opened && pendingEvents === 0 && !config.revalidate) {
           enqueue(payload)
           return
         }
@@ -180,31 +184,41 @@ export function createSSEStream(request: NextRequest, config: SSEStreamConfig):
           return
         }
         pendingEvents += 1
+        // Authorized as soon as it arrives, so concurrent events share one check, but written only
+        // after every event before it.
         const authorized = opened
           ? revalidate()
-          : opening.then(() => (cleaned ? undefined : revalidate()))
-        void authorized.then(
-          () => {
-            pendingEvents -= 1
-            enqueue(payload)
-          },
-          () => {
-            pendingEvents -= 1
-            close('authorization_lost')
-          }
-        )
+          : writes.then(() => (cleaned ? undefined : revalidate()))
+        authorized.catch(noop)
+        writes = writes
+          .then(() => authorized)
+          .then(
+            () => {
+              pendingEvents -= 1
+              enqueue(payload)
+            },
+            () => {
+              pendingEvents -= 1
+              close('authorization_lost')
+            }
+          )
       }
 
       try {
         // The runtime sends the status and headers with the first body chunk, so the stream writes
         // one as soon as it opens. A client reads its state once the stream opens, so it opens once
         // every subscription receives events, or at the deadline when one is in an outage. A
         // heartbeat cannot open it first: the deadline is shorter than the heartbeat interval.
-        opening = Promise.race([
+        // Closing settles it too, so a stream closed before it opened is not kept alive by a
+        // subscription that never becomes ready.
+        writes = Promise.race([
           Promise.all(config.subscriptions.map((subscription) => subscription.ready?.())),
           new Promise<void>((resolve) => {
             const deadline = setTimeout(resolve, OPEN_DEADLINE_MS)
-            teardowns.push(() => clearTimeout(deadline))
+            teardowns.push(() => {
+              clearTimeout(deadline)
+              resolve()
+            })
           }),
         ]).then(
           () => {