From fb1fead94d6d8a4b0d9e37b622168d6415b6bd99 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Mon, 24 Aug 2026 19:44:17 -0700 Subject: [PATCH 1/5] fix(sse): bound workspace SSE connection lifetime Teardown ran only from the request abort listener and the stream cancel callback, both of which fire only when the runtime reports a client disconnect. Nothing else bounded the connection, so a missed report left the pub/sub handler, the heartbeat timer, and the stream's undrained queue held for the life of the process. Add a jittered lifetime ceiling checked on the existing heartbeat tick, tighten reclaim for a vanished consumer via desiredSize, remove the abort listener on every teardown path, and run full teardown when a heartbeat enqueue fails. Log the close reason so opens minus closes is observable. --- apps/sim/lib/events/sse-endpoint.test.ts | 111 +++++++++++++++++++++++ apps/sim/lib/events/sse-endpoint.ts | 94 ++++++++++++++----- 2 files changed, 180 insertions(+), 25 deletions(-) create mode 100644 apps/sim/lib/events/sse-endpoint.test.ts diff --git a/apps/sim/lib/events/sse-endpoint.test.ts b/apps/sim/lib/events/sse-endpoint.test.ts new file mode 100644 index 00000000000..242a9be1b7c --- /dev/null +++ b/apps/sim/lib/events/sse-endpoint.test.ts @@ -0,0 +1,111 @@ +/** + * @vitest-environment node + */ + +import { authMockFns, permissionsMock, permissionsMockFns } from '@sim/testing' +import { NextRequest } from 'next/server' +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { + createWorkspaceSSE, + HEARTBEAT_INTERVAL_MS, + MAX_CONNECTION_JITTER_MS, + MAX_CONNECTION_MS, + MAX_UNDRAINED_CHUNKS, +} from '@/lib/events/sse-endpoint' + +vi.mock('@/lib/workspaces/permissions/utils', () => permissionsMock) + +const PAST_CEILING_MS = MAX_CONNECTION_MS + MAX_CONNECTION_JITTER_MS + HEARTBEAT_INTERVAL_MS + +/** Enough undrained heartbeats to trip the unread check, and no more. */ +const PAST_UNREAD_MS = (MAX_UNDRAINED_CHUNKS + 2) * HEARTBEAT_INTERVAL_MS + +async function openConnection(signal: AbortSignal = new AbortController().signal) { + const unsubscribe = vi.fn() + const handler = createWorkspaceSSE({ + label: 'test', + subscriptions: [{ subscribe: () => unsubscribe }], + }) + const request = new NextRequest(new URL('https://sim.test/api/test/events?workspaceId=ws-1'), { + signal, + }) + const response = await handler(request) + + return { body: response.body as ReadableStream, unsubscribe } +} + +/** Resolves once the stream closes. */ +async function drain(body: ReadableStream): Promise { + const reader = body.getReader() + while (true) { + const { done } = await reader.read() + if (done) return + } +} + +describe('createWorkspaceSSE', () => { + beforeEach(() => { + vi.clearAllMocks() + vi.useFakeTimers() + authMockFns.mockGetSession.mockResolvedValue({ user: { id: 'user-1' } }) + permissionsMockFns.mockGetUserEntityPermissions.mockResolvedValue('admin') + }) + + afterEach(() => { + vi.useRealTimers() + }) + + it('releases subscriptions and closes the stream when the connection reaches its ceiling', async () => { + const { body, unsubscribe } = await openConnection() + const drained = drain(body) + + await vi.advanceTimersByTimeAsync(PAST_CEILING_MS) + + await drained + expect(unsubscribe).toHaveBeenCalledTimes(1) + }) + + it('releases subscriptions when the consumer stops draining the stream', async () => { + const { unsubscribe } = await openConnection() + + await vi.advanceTimersByTimeAsync(PAST_UNREAD_MS) + + expect(unsubscribe).toHaveBeenCalledTimes(1) + }) + + it('keeps a drained connection alive past the unread threshold', async () => { + const { body, unsubscribe } = await openConnection() + void drain(body) + + await vi.advanceTimersByTimeAsync(PAST_UNREAD_MS) + + expect(unsubscribe).not.toHaveBeenCalled() + }) + + it('releases subscriptions when the request aborts', async () => { + const controller = new AbortController() + const { unsubscribe } = await openConnection(controller.signal) + + controller.abort() + + expect(unsubscribe).toHaveBeenCalledTimes(1) + }) + + it('releases subscriptions when the consumer cancels the stream', async () => { + const { body, unsubscribe } = await openConnection() + + await body.cancel() + + expect(unsubscribe).toHaveBeenCalledTimes(1) + }) + + it('releases subscriptions once when abort and the ceiling both elapse', async () => { + const controller = new AbortController() + const { unsubscribe } = await openConnection(controller.signal) + + controller.abort() + await vi.advanceTimersByTimeAsync(PAST_CEILING_MS) + + expect(unsubscribe).toHaveBeenCalledTimes(1) + }) +}) diff --git a/apps/sim/lib/events/sse-endpoint.ts b/apps/sim/lib/events/sse-endpoint.ts index 30cc46619e0..ec012f603ba 100644 --- a/apps/sim/lib/events/sse-endpoint.ts +++ b/apps/sim/lib/events/sse-endpoint.ts @@ -6,6 +6,7 @@ */ import { createLogger } from '@sim/logger' +import { randomFloat } from '@sim/utils/random' import type { NextRequest } from 'next/server' import { getSession } from '@/lib/auth' import { SSE_HEADERS } from '@/lib/core/utils/sse' @@ -23,7 +24,34 @@ interface WorkspaceSSEConfig { subscriptions: SSESubscription[] } -const HEARTBEAT_INTERVAL_MS = 30_000 +const encoder = new TextEncoder() + +export const HEARTBEAT_INTERVAL_MS = 30_000 + +/** + * Hard ceiling on one connection's lifetime. + * + * `request.signal` abort and stream `cancel()` are the primary teardown paths, + * but both fire only when the runtime reports the client disconnect. This + * ceiling releases the connection without depending on that report, so a + * missed disconnect costs one connection rather than accumulating for the life + * of the process. `EventSource` reconnects on its own, so delivery continues + * across the boundary. + */ +export const MAX_CONNECTION_MS = 15 * 60 * 1000 + +/** Spreads reconnects so connections opened together do not expire together. */ +export const MAX_CONNECTION_JITTER_MS = 60_000 + +/** + * Undrained chunk count that marks a connection unread, which shortens the + * reclaim window for a vanished consumer that the ceiling above would + * otherwise hold for its full duration. The default queuing strategy reports + * `desiredSize` as `1 - queued`, so this trips only once the consumer has + * pulled nothing for several minutes — well beyond the transient backpressure + * of a slow but live client. + */ +export const MAX_UNDRAINED_CHUNKS = 16 export function createWorkspaceSSE(config: WorkspaceSSEConfig) { const logger = createLogger(`${config.label}-SSE`) @@ -45,21 +73,30 @@ export function createWorkspaceSSE(config: WorkspaceSSEConfig) { return new Response('Access denied to workspace', { status: 403 }) } - const encoder = new TextEncoder() - const unsubscribers: Array<() => void> = [] + const teardowns: Array<() => void> = [] let cleaned = false - const cleanup = () => { + const cleanup = (reason: string) => { if (cleaned) return cleaned = true - for (const unsub of unsubscribers) { - unsub() + for (const teardown of teardowns) { + teardown() } - logger.info(`SSE connection closed for workspace ${workspaceId}`) + teardowns.length = 0 + logger.info(`SSE connection closed for workspace ${workspaceId}`, { reason }) } const stream = new ReadableStream({ start(controller) { + const close = (reason: string) => { + cleanup(reason) + try { + controller.close() + } catch { + // Already closed + } + } + const send = (eventName: string, data: Record) => { if (cleaned) return try { @@ -72,40 +109,47 @@ export function createWorkspaceSSE(config: WorkspaceSSEConfig) { } for (const subscription of config.subscriptions) { - const unsub = subscription.subscribe(workspaceId, send) - unsubscribers.push(unsub) + teardowns.push(subscription.subscribe(workspaceId, send)) } + const deadline = Date.now() + MAX_CONNECTION_MS + randomFloat() * MAX_CONNECTION_JITTER_MS + const heartbeat = setInterval(() => { if (cleaned) { clearInterval(heartbeat) return } + if (Date.now() >= deadline) { + close('expired') + return + } + const desiredSize = controller.desiredSize + if (desiredSize !== null && desiredSize <= -MAX_UNDRAINED_CHUNKS) { + close('unread') + return + } try { controller.enqueue(encoder.encode(': heartbeat\n\n')) } catch { - clearInterval(heartbeat) + close('errored') } }, HEARTBEAT_INTERVAL_MS) - unsubscribers.push(() => clearInterval(heartbeat)) - - request.signal.addEventListener( - 'abort', - () => { - cleanup() - try { - controller.close() - } catch { - // Already closed - } - }, - { once: true } - ) + teardowns.push(() => clearInterval(heartbeat)) + + // `once` only self-removes if abort fires; the expiry and unread paths + // close the connection while the signal is still live, so the listener + // needs its own removal or it retains this whole scope. + const listenerScope = new AbortController() + request.signal.addEventListener('abort', () => close('aborted'), { + once: true, + signal: listenerScope.signal, + }) + teardowns.push(() => listenerScope.abort()) logger.info(`SSE connection opened for workspace ${workspaceId}`) }, cancel() { - cleanup() + cleanup('cancelled') }, }) From 0c22829f66919d65007fd90b8f9735ed00b3353c Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Mon, 24 Aug 2026 19:58:13 -0700 Subject: [PATCH 2/5] fix(mothership): resync chat caches after an SSE reconnect gap task_status events are transient and never replayed, so any window with no open connection can drop a create, rename, delete, or completion. The chat hook reconnected silently and reconciled nothing, leaving list and detail caches stale until an unrelated action refreshed them. Resync on reconnect, on the first open of a re-subscription, and on a first open that only succeeded after an error, matching the pattern useMcpToolsEvents already uses for the same gap. --- .../hooks/use-mothership-chat-events.test.ts | 27 ++++++++++++- apps/sim/hooks/use-mothership-chat-events.ts | 40 +++++++++++++++++++ 2 files changed, 66 insertions(+), 1 deletion(-) diff --git a/apps/sim/hooks/use-mothership-chat-events.test.ts b/apps/sim/hooks/use-mothership-chat-events.test.ts index 52475b0ecf6..89b3c2d7fdb 100644 --- a/apps/sim/hooks/use-mothership-chat-events.test.ts +++ b/apps/sim/hooks/use-mothership-chat-events.test.ts @@ -14,7 +14,10 @@ vi.mock('@/lib/browser-agent/transport', () => ({ suspendBrowserScope })) vi.mock('@/lib/terminal/transport', () => ({ suspendTerminalScope })) import { mothershipChatKeys } from '@/hooks/queries/mothership-chats' -import { handleMothershipChatStatusEvent } from '@/hooks/use-mothership-chat-events' +import { + handleMothershipChatStatusEvent, + resyncMothershipChatCaches, +} from '@/hooks/use-mothership-chat-events' describe('handleMothershipChatStatusEvent', () => { const queryClient = { @@ -419,3 +422,25 @@ describe('handleMothershipChatStatusEvent', () => { expect(queryClient.removeQueries).not.toHaveBeenCalled() }) }) + +describe('resyncMothershipChatCaches', () => { + const queryClient = { + invalidateQueries: vi.fn().mockResolvedValue(undefined), + } satisfies Pick + + beforeEach(() => { + vi.clearAllMocks() + }) + + it('invalidates the workspace lists and every chat detail', () => { + resyncMothershipChatCaches(queryClient, 'ws-1') + + expect(queryClient.invalidateQueries).toHaveBeenCalledTimes(2) + expect(queryClient.invalidateQueries).toHaveBeenCalledWith({ + queryKey: mothershipChatKeys.workspaceLists('ws-1'), + }) + expect(queryClient.invalidateQueries).toHaveBeenCalledWith({ + queryKey: mothershipChatKeys.details(), + }) + }) +}) diff --git a/apps/sim/hooks/use-mothership-chat-events.ts b/apps/sim/hooks/use-mothership-chat-events.ts index 096aaed5812..e0ac3cc2245 100644 --- a/apps/sim/hooks/use-mothership-chat-events.ts +++ b/apps/sim/hooks/use-mothership-chat-events.ts @@ -9,6 +9,9 @@ import { type MothershipChatHistory, mothershipChatKeys } from '@/hooks/queries/ const logger = createLogger('MothershipChatEvents') +/** Workspaces this process has subscribed to before, so a re-subscribe can be told from a first one. */ +const everSubscribed = new Set() + const CHAT_STATUS_TYPES = ['started', 'completed', 'created', 'deleted', 'renamed'] as const type ChatStatusEventType = (typeof CHAT_STATUS_TYPES)[number] const CHAT_STATUS_TYPE_SET = new Set(CHAT_STATUS_TYPES) @@ -128,6 +131,22 @@ export function handleMothershipChatStatusEvent( } } +/** + * Re-syncs chat caches after a gap in the event stream. + * + * `task_status` events are transient — nothing replays what was published while + * no connection was open — so any reconnect may have missed a create, rename, + * delete, or completion. Invalidating the workspace lists and every chat detail + * reconciles from the server; only queries that are currently mounted refetch. + */ +export function resyncMothershipChatCaches( + queryClient: Pick, + workspaceId: string +): void { + queryClient.invalidateQueries({ queryKey: mothershipChatKeys.workspaceLists(workspaceId) }) + queryClient.invalidateQueries({ queryKey: mothershipChatKeys.details() }) +} + /** * Subscribes to chat status SSE events and invalidates chat caches on changes. * The SSE event name remains `task_status` for wire compatibility. @@ -154,7 +173,28 @@ export function useMothershipChatEvents(workspaceId: string | undefined) { ) }) + // `onopen` fires on the initial connect and on every auto-reconnect. Re-sync + // whenever a gap could have swallowed an event: on any reconnect, on the + // first open of a RE-subscription (switching workspace away and back tears + // the connection down, and the list/detail queries remount inside their + // stale times so they do not refetch on their own), and on a first open that + // only succeeded after an error (the initial queries may have failed during + // that gap and will not retry themselves). Skip only a clean first + // subscription — those queries fetch fresh on their own initial mount. + const isResubscribe = everSubscribed.has(workspaceId) + everSubscribed.add(workspaceId) + let opened = false + let erroredBeforeOpen = false + + eventSource.onopen = () => { + if (opened || isResubscribe || erroredBeforeOpen) { + resyncMothershipChatCaches(queryClient, workspaceId) + } + opened = true + } + eventSource.onerror = () => { + if (!opened) erroredBeforeOpen = true logger.warn(`SSE connection error for workspace ${workspaceId}`) } From 70541f43c009417cd276d9bf33af2105890df689 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Mon, 24 Aug 2026 20:11:33 -0700 Subject: [PATCH 3/5] fix(mothership): keep reconnect resync off locally streaming chats The resync invalidated every chat detail, including one whose stream this client is rendering optimistically. Refetching there replaces the local transcript with a server copy that does not yet hold the in-flight message, which is exactly what status events avoid via shouldSkipDetailInvalidationForStreamEvent. Filter the detail invalidation with the same isLocalOptimisticActiveStream check. Those chats reconcile when their own stream finishes. --- .../hooks/use-mothership-chat-events.test.ts | 34 ++++++++++++++++--- apps/sim/hooks/use-mothership-chat-events.ts | 11 +++++- 2 files changed, 40 insertions(+), 5 deletions(-) diff --git a/apps/sim/hooks/use-mothership-chat-events.test.ts b/apps/sim/hooks/use-mothership-chat-events.test.ts index 89b3c2d7fdb..2222af48e80 100644 --- a/apps/sim/hooks/use-mothership-chat-events.test.ts +++ b/apps/sim/hooks/use-mothership-chat-events.test.ts @@ -432,15 +432,41 @@ describe('resyncMothershipChatCaches', () => { vi.clearAllMocks() }) - it('invalidates the workspace lists and every chat detail', () => { + function detailPredicate() { + resyncMothershipChatCaches(queryClient, 'ws-1') + const predicate = queryClient.invalidateQueries.mock.calls + .map(([arg]: [{ predicate?: (query: unknown) => boolean }]) => arg.predicate) + .find(Boolean) + if (!predicate) throw new Error('detail invalidation did not pass a predicate') + return (data: unknown) => predicate({ state: { data } }) + } + + it('invalidates the workspace lists and the chat details', () => { resyncMothershipChatCaches(queryClient, 'ws-1') expect(queryClient.invalidateQueries).toHaveBeenCalledTimes(2) expect(queryClient.invalidateQueries).toHaveBeenCalledWith({ queryKey: mothershipChatKeys.workspaceLists('ws-1'), }) - expect(queryClient.invalidateQueries).toHaveBeenCalledWith({ - queryKey: mothershipChatKeys.details(), - }) + expect(queryClient.invalidateQueries).toHaveBeenCalledWith( + expect.objectContaining({ queryKey: mothershipChatKeys.details() }) + ) + }) + + it('skips the detail of a chat whose stream this client is rendering locally', () => { + expect( + detailPredicate()({ + messages: [{ id: 'new-stream' }, { id: 'live-assistant:new-stream' }], + activeStreamId: 'new-stream', + }) + ).toBe(false) + }) + + it('invalidates details with no active stream, and streams not rendered locally', () => { + const predicate = detailPredicate() + + expect(predicate(undefined)).toBe(true) + expect(predicate({ messages: [{ id: 'stream-1' }] })).toBe(true) + expect(predicate({ messages: [{ id: 'stream-1' }], activeStreamId: 'stream-1' })).toBe(true) }) }) diff --git a/apps/sim/hooks/use-mothership-chat-events.ts b/apps/sim/hooks/use-mothership-chat-events.ts index e0ac3cc2245..bf67697e4f5 100644 --- a/apps/sim/hooks/use-mothership-chat-events.ts +++ b/apps/sim/hooks/use-mothership-chat-events.ts @@ -138,13 +138,22 @@ export function handleMothershipChatStatusEvent( * no connection was open — so any reconnect may have missed a create, rename, * delete, or completion. Invalidating the workspace lists and every chat detail * reconciles from the server; only queries that are currently mounted refetch. + * + * A chat whose stream this client is rendering locally is left alone, for the + * same reason status events skip it: refetching mid-stream would replace the + * optimistic transcript with a server copy that does not yet hold the in-flight + * message. That chat reconciles when its own stream finishes. */ export function resyncMothershipChatCaches( queryClient: Pick, workspaceId: string ): void { queryClient.invalidateQueries({ queryKey: mothershipChatKeys.workspaceLists(workspaceId) }) - queryClient.invalidateQueries({ queryKey: mothershipChatKeys.details() }) + queryClient.invalidateQueries({ + queryKey: mothershipChatKeys.details(), + predicate: (query) => + !isLocalOptimisticActiveStream(query.state.data as MothershipChatHistory | undefined), + }) } /** From f76765fbc15dcdabfd899deea20d866557f79a73 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Mon, 24 Aug 2026 20:27:28 -0700 Subject: [PATCH 4/5] fix(mothership): only skip resync for a stream still running MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Optimistic markers alone were the skip condition, but a finished turn can leave activeStreamId and its live-assistant message cached when finalization skips detail invalidation for a queued follow-up. That chat would then be excluded from every future resync — permanently, since only a refetch clears the markers, and the resync was the refetch. Gate the skip on a non-terminal streamSnapshot status so it covers turns that are genuinely still streaming. Exports isTerminalStreamStatus, which was already the private check for this in effective-transcript. --- .../hooks/use-mothership-chat-events.test.ts | 16 +++++++++- apps/sim/hooks/use-mothership-chat-events.ts | 32 +++++++++++++++---- .../lib/copilot/chat/effective-transcript.ts | 2 +- 3 files changed, 42 insertions(+), 8 deletions(-) diff --git a/apps/sim/hooks/use-mothership-chat-events.test.ts b/apps/sim/hooks/use-mothership-chat-events.test.ts index 2222af48e80..398483744e7 100644 --- a/apps/sim/hooks/use-mothership-chat-events.test.ts +++ b/apps/sim/hooks/use-mothership-chat-events.test.ts @@ -453,15 +453,29 @@ describe('resyncMothershipChatCaches', () => { ) }) - it('skips the detail of a chat whose stream this client is rendering locally', () => { + it('skips the detail of a chat this client is still streaming', () => { expect( detailPredicate()({ messages: [{ id: 'new-stream' }, { id: 'live-assistant:new-stream' }], activeStreamId: 'new-stream', + streamSnapshot: { events: [], previewSessions: [], status: 'streaming' }, }) ).toBe(false) }) + it('invalidates a chat whose stream finished but left its optimistic markers cached', () => { + const predicate = detailPredicate() + const finished = (status: string) => ({ + messages: [{ id: 'new-stream' }, { id: 'live-assistant:new-stream' }], + activeStreamId: 'new-stream', + streamSnapshot: { events: [], previewSessions: [], status }, + }) + + expect(predicate(finished('complete'))).toBe(true) + expect(predicate(finished('error'))).toBe(true) + expect(predicate(finished('cancelled'))).toBe(true) + }) + it('invalidates details with no active stream, and streams not rendered locally', () => { const predicate = detailPredicate() diff --git a/apps/sim/hooks/use-mothership-chat-events.ts b/apps/sim/hooks/use-mothership-chat-events.ts index bf67697e4f5..f5633cc3463 100644 --- a/apps/sim/hooks/use-mothership-chat-events.ts +++ b/apps/sim/hooks/use-mothership-chat-events.ts @@ -2,7 +2,10 @@ import { useEffect } from 'react' import { createLogger } from '@sim/logger' import type { QueryClient } from '@tanstack/react-query' import { useQueryClient } from '@tanstack/react-query' -import { getLiveAssistantMessageId } from '@/lib/copilot/chat/effective-transcript' +import { + getLiveAssistantMessageId, + isTerminalStreamStatus, +} from '@/lib/copilot/chat/effective-transcript' import { isChatEnabled } from '@/lib/core/config/env-flags' import { suspendDesktopChatScopes } from '@/lib/desktop/chat-scope' import { type MothershipChatHistory, mothershipChatKeys } from '@/hooks/queries/mothership-chats' @@ -38,6 +41,23 @@ function isLocalOptimisticActiveStream(current: MothershipChatHistory | undefine return current.messages.some((message) => message.id === liveAssistantId) } +/** + * True while this client is still rendering a stream for the chat. + * + * The optimistic markers alone are not enough: a finished turn can leave + * `activeStreamId` and its live-assistant message in the cache (finalization + * skips detail invalidation when a follow-up is queued), and treating those as + * live would exclude the chat from every future resync — permanently, since + * only a refetch would clear them. Requiring a non-terminal snapshot status + * keeps the skip to turns that are genuinely still streaming. + */ +function isStreamingLocally(current: MothershipChatHistory | undefined) { + return ( + isLocalOptimisticActiveStream(current) && + !isTerminalStreamStatus(current?.streamSnapshot?.status) + ) +} + /** * Returns true when the cached active stream is known to be later in the * chronological transcript than the stream that emitted this status event. @@ -139,10 +159,10 @@ export function handleMothershipChatStatusEvent( * delete, or completion. Invalidating the workspace lists and every chat detail * reconciles from the server; only queries that are currently mounted refetch. * - * A chat whose stream this client is rendering locally is left alone, for the - * same reason status events skip it: refetching mid-stream would replace the - * optimistic transcript with a server copy that does not yet hold the in-flight - * message. That chat reconciles when its own stream finishes. + * A chat this client is still streaming is left alone, for the same reason + * status events skip it: refetching mid-stream would replace the optimistic + * transcript with a server copy that does not yet hold the in-flight message. + * That chat reconciles when its own stream finishes. */ export function resyncMothershipChatCaches( queryClient: Pick, @@ -152,7 +172,7 @@ export function resyncMothershipChatCaches( queryClient.invalidateQueries({ queryKey: mothershipChatKeys.details(), predicate: (query) => - !isLocalOptimisticActiveStream(query.state.data as MothershipChatHistory | undefined), + !isStreamingLocally(query.state.data as MothershipChatHistory | undefined), }) } diff --git a/apps/sim/lib/copilot/chat/effective-transcript.ts b/apps/sim/lib/copilot/chat/effective-transcript.ts index fd1e4423590..3f3c7d9f1b1 100644 --- a/apps/sim/lib/copilot/chat/effective-transcript.ts +++ b/apps/sim/lib/copilot/chat/effective-transcript.ts @@ -52,7 +52,7 @@ function asPayloadRecord(value: unknown): Record | undefined { return isRecordLike(value) ? value : undefined } -function isTerminalStreamStatus(status: string | null | undefined): boolean { +export function isTerminalStreamStatus(status: string | null | undefined): boolean { return ( status === MothershipStreamV1CompletionStatus.complete || status === MothershipStreamV1CompletionStatus.error || From cbe889839daf831833705e3ad60b98bd7c0f82f5 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Mon, 24 Aug 2026 20:34:54 -0700 Subject: [PATCH 5/5] fix(sse): raise the ceiling and narrow reconnect resync to the lists MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Deciding from cache whether a chat is still streaming is not reliable — the optimistic markers outlive the turn, and each refinement of that predicate exposed another state where it answers wrongly. Drop it: the resync now invalidates only the workspace lists, which is always safe, and chat detail reconciliation stays as it is today rather than being half-solved here. Raise the ceiling to 4h, matching lib/realtime/event-stream-route.ts. A healthy client is drained and so is never unread; the unread check is what reclaims a vanished consumer, and it does so within minutes. A short ceiling would therefore only force reconnects on the connections that are working, and every reconnect is a window where a transient event can be missed. Retention stays bounded by the ceiling instead of by process uptime. --- .../hooks/use-mothership-chat-events.test.ts | 49 +++---------------- apps/sim/hooks/use-mothership-chat-events.ts | 45 +++++------------ .../lib/copilot/chat/effective-transcript.ts | 2 +- apps/sim/lib/events/sse-endpoint.ts | 20 +++++--- 4 files changed, 32 insertions(+), 84 deletions(-) diff --git a/apps/sim/hooks/use-mothership-chat-events.test.ts b/apps/sim/hooks/use-mothership-chat-events.test.ts index 398483744e7..fb01186b06c 100644 --- a/apps/sim/hooks/use-mothership-chat-events.test.ts +++ b/apps/sim/hooks/use-mothership-chat-events.test.ts @@ -432,55 +432,20 @@ describe('resyncMothershipChatCaches', () => { vi.clearAllMocks() }) - function detailPredicate() { - resyncMothershipChatCaches(queryClient, 'ws-1') - const predicate = queryClient.invalidateQueries.mock.calls - .map(([arg]: [{ predicate?: (query: unknown) => boolean }]) => arg.predicate) - .find(Boolean) - if (!predicate) throw new Error('detail invalidation did not pass a predicate') - return (data: unknown) => predicate({ state: { data } }) - } - - it('invalidates the workspace lists and the chat details', () => { + it('invalidates the workspace lists', () => { resyncMothershipChatCaches(queryClient, 'ws-1') - expect(queryClient.invalidateQueries).toHaveBeenCalledTimes(2) + expect(queryClient.invalidateQueries).toHaveBeenCalledTimes(1) expect(queryClient.invalidateQueries).toHaveBeenCalledWith({ queryKey: mothershipChatKeys.workspaceLists('ws-1'), }) - expect(queryClient.invalidateQueries).toHaveBeenCalledWith( - expect.objectContaining({ queryKey: mothershipChatKeys.details() }) - ) }) - it('skips the detail of a chat this client is still streaming', () => { - expect( - detailPredicate()({ - messages: [{ id: 'new-stream' }, { id: 'live-assistant:new-stream' }], - activeStreamId: 'new-stream', - streamSnapshot: { events: [], previewSessions: [], status: 'streaming' }, - }) - ).toBe(false) - }) - - it('invalidates a chat whose stream finished but left its optimistic markers cached', () => { - const predicate = detailPredicate() - const finished = (status: string) => ({ - messages: [{ id: 'new-stream' }, { id: 'live-assistant:new-stream' }], - activeStreamId: 'new-stream', - streamSnapshot: { events: [], previewSessions: [], status }, - }) - - expect(predicate(finished('complete'))).toBe(true) - expect(predicate(finished('error'))).toBe(true) - expect(predicate(finished('cancelled'))).toBe(true) - }) - - it('invalidates details with no active stream, and streams not rendered locally', () => { - const predicate = detailPredicate() + it('leaves chat details untouched so a mounted stream cannot be refetched mid-turn', () => { + resyncMothershipChatCaches(queryClient, 'ws-1') - expect(predicate(undefined)).toBe(true) - expect(predicate({ messages: [{ id: 'stream-1' }] })).toBe(true) - expect(predicate({ messages: [{ id: 'stream-1' }], activeStreamId: 'stream-1' })).toBe(true) + expect(queryClient.invalidateQueries).not.toHaveBeenCalledWith( + expect.objectContaining({ queryKey: mothershipChatKeys.details() }) + ) }) }) diff --git a/apps/sim/hooks/use-mothership-chat-events.ts b/apps/sim/hooks/use-mothership-chat-events.ts index f5633cc3463..175891a4dbd 100644 --- a/apps/sim/hooks/use-mothership-chat-events.ts +++ b/apps/sim/hooks/use-mothership-chat-events.ts @@ -2,10 +2,7 @@ import { useEffect } from 'react' import { createLogger } from '@sim/logger' import type { QueryClient } from '@tanstack/react-query' import { useQueryClient } from '@tanstack/react-query' -import { - getLiveAssistantMessageId, - isTerminalStreamStatus, -} from '@/lib/copilot/chat/effective-transcript' +import { getLiveAssistantMessageId } from '@/lib/copilot/chat/effective-transcript' import { isChatEnabled } from '@/lib/core/config/env-flags' import { suspendDesktopChatScopes } from '@/lib/desktop/chat-scope' import { type MothershipChatHistory, mothershipChatKeys } from '@/hooks/queries/mothership-chats' @@ -41,23 +38,6 @@ function isLocalOptimisticActiveStream(current: MothershipChatHistory | undefine return current.messages.some((message) => message.id === liveAssistantId) } -/** - * True while this client is still rendering a stream for the chat. - * - * The optimistic markers alone are not enough: a finished turn can leave - * `activeStreamId` and its live-assistant message in the cache (finalization - * skips detail invalidation when a follow-up is queued), and treating those as - * live would exclude the chat from every future resync — permanently, since - * only a refetch would clear them. Requiring a non-terminal snapshot status - * keeps the skip to turns that are genuinely still streaming. - */ -function isStreamingLocally(current: MothershipChatHistory | undefined) { - return ( - isLocalOptimisticActiveStream(current) && - !isTerminalStreamStatus(current?.streamSnapshot?.status) - ) -} - /** * Returns true when the cached active stream is known to be later in the * chronological transcript than the stream that emitted this status event. @@ -152,28 +132,25 @@ export function handleMothershipChatStatusEvent( } /** - * Re-syncs chat caches after a gap in the event stream. + * Re-syncs the workspace chat lists after a gap in the event stream. * * `task_status` events are transient — nothing replays what was published while - * no connection was open — so any reconnect may have missed a create, rename, - * delete, or completion. Invalidating the workspace lists and every chat detail - * reconciles from the server; only queries that are currently mounted refetch. + * no connection was open — so a reconnect may have missed a create, rename, or + * delete. The lists carry that workspace-level state, and refetching them is + * always safe. * - * A chat this client is still streaming is left alone, for the same reason - * status events skip it: refetching mid-stream would replace the optimistic - * transcript with a server copy that does not yet hold the in-flight message. - * That chat reconciles when its own stream finishes. + * Chat details are deliberately left alone. The only one that would refetch is + * the mounted chat, which may be rendering an in-flight stream, and refetching + * there replaces the optimistic transcript with a server copy that does not yet + * hold the streaming message. Cached state cannot reliably say whether a turn is + * still running — the optimistic markers outlive it — so detail reconciliation + * stays as it is today and belongs with the streaming state that can answer it. */ export function resyncMothershipChatCaches( queryClient: Pick, workspaceId: string ): void { queryClient.invalidateQueries({ queryKey: mothershipChatKeys.workspaceLists(workspaceId) }) - queryClient.invalidateQueries({ - queryKey: mothershipChatKeys.details(), - predicate: (query) => - !isStreamingLocally(query.state.data as MothershipChatHistory | undefined), - }) } /** diff --git a/apps/sim/lib/copilot/chat/effective-transcript.ts b/apps/sim/lib/copilot/chat/effective-transcript.ts index 3f3c7d9f1b1..fd1e4423590 100644 --- a/apps/sim/lib/copilot/chat/effective-transcript.ts +++ b/apps/sim/lib/copilot/chat/effective-transcript.ts @@ -52,7 +52,7 @@ function asPayloadRecord(value: unknown): Record | undefined { return isRecordLike(value) ? value : undefined } -export function isTerminalStreamStatus(status: string | null | undefined): boolean { +function isTerminalStreamStatus(status: string | null | undefined): boolean { return ( status === MothershipStreamV1CompletionStatus.complete || status === MothershipStreamV1CompletionStatus.error || diff --git a/apps/sim/lib/events/sse-endpoint.ts b/apps/sim/lib/events/sse-endpoint.ts index ec012f603ba..8e5df184e1d 100644 --- a/apps/sim/lib/events/sse-endpoint.ts +++ b/apps/sim/lib/events/sse-endpoint.ts @@ -29,16 +29,22 @@ const encoder = new TextEncoder() export const HEARTBEAT_INTERVAL_MS = 30_000 /** - * Hard ceiling on one connection's lifetime. + * Defensive ceiling on one connection's lifetime; `EventSource` reconnects past + * this, so delivery continues across the boundary. * * `request.signal` abort and stream `cancel()` are the primary teardown paths, - * but both fire only when the runtime reports the client disconnect. This - * ceiling releases the connection without depending on that report, so a - * missed disconnect costs one connection rather than accumulating for the life - * of the process. `EventSource` reconnects on its own, so delivery continues - * across the boundary. + * but both fire only when the runtime reports the client disconnect, and the + * unread check below only catches a consumer that has stopped draining. This + * releases whatever both miss, so retention is bounded by the ceiling instead + * of by process uptime. + * + * Matches the ceiling `lib/realtime/event-stream-route.ts` already uses for the + * same purpose. It is deliberately far longer than the unread window: a healthy + * client is drained and therefore never unread, so a short ceiling would only + * force reconnects on the connections that are working, and every reconnect is + * a window in which a transient event can be missed. */ -export const MAX_CONNECTION_MS = 15 * 60 * 1000 +export const MAX_CONNECTION_MS = 4 * 60 * 60 * 1000 /** Spreads reconnects so connections opened together do not expire together. */ export const MAX_CONNECTION_JITTER_MS = 60_000