Skip to content

Commit 3c266bf

Browse files
committed
fix(mothership): close a pre-aborted SSE stream, keep the new-chat effort across a failed first send
- createSSEStream closes at once when the request aborted before start(), instead of subscribing until rotation. - The new-chat effort pick is dropped by useChat when its chatless surface is left, not by each composer's unmount, so a failed first send keeps it and the pending chat view shows it. - The chat response's effort is optional, so a new client loads chats from a server that predates it.
1 parent 8760ca7 commit 3c266bf

8 files changed

Lines changed: 145 additions & 10 deletions

File tree

‎apps/sim/app/workspace/[workspaceId]/home/components/user-input/components/model-selector.tsx‎

Lines changed: 0 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,5 @@
11
'use client'
22

3-
import { useEffect } from 'react'
43
import {
54
DropdownMenu,
65
DropdownMenuContent,
@@ -53,11 +52,6 @@ export function ModelSelector() {
5352
else setNewChatEffort(choice)
5453
}
5554

56-
useEffect(() => {
57-
if (chatId) return
58-
return () => useMothershipEffortStore.getState().setNewChatEffort(null)
59-
}, [chatId])
60-
6155
const effortLabel = options.find((option) => option.value === effort)?.label ?? effort
6256

6357
return (

‎apps/sim/app/workspace/[workspaceId]/home/hooks/use-chat.dom.test.tsx‎

Lines changed: 89 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -92,6 +92,7 @@ import { MOTHERSHIP_STREAM_REPLAY_HEADER } from '@/lib/mothership/constants'
9292
import type { MothershipStreamV1EventEnvelope } from '@/lib/mothership/generated/mothership-stream-v1'
9393
import { getChatResourceSelectionId } from '@/lib/mothership/resources/types'
9494
import { collectCitedMessageSources } from '@/app/workspace/[workspaceId]/home/components/message-content/message-sources'
95+
import { ModelSelector } from '@/app/workspace/[workspaceId]/home/components/user-input/components/model-selector'
9596
import {
9697
readQueuedSendHandoffState,
9798
writeQueuedSendHandoffState,
@@ -4701,6 +4702,94 @@ describe('useChat remount send recovery', () => {
47014702
])
47024703
})
47034704

4705+
it('keeps the new-chat effort across the composer swap of a first send that fails', async () => {
4706+
useMothershipEffortStore.getState().reset()
4707+
const post = Promise.withResolvers<Response>()
4708+
vi.stubGlobal('fetch', (input: RequestInfo | URL, init?: RequestInit) => {
4709+
if (String(input) !== '/api/mothership/chat' || init?.method !== 'POST') {
4710+
return fetchStub(input, init)
4711+
}
4712+
state.postBodies.push(JSON.parse(String(init.body)))
4713+
return post.promise
4714+
})
4715+
;(globalThis as { IS_REACT_ACT_ENVIRONMENT?: boolean }).IS_REACT_ACT_ENVIRONMENT = true
4716+
queryClient = new QueryClient({ defaultOptions: { queries: { retry: false } } })
4717+
const container = document.createElement('div')
4718+
const root = createRoot(container)
4719+
mountedRoots.push(root)
4720+
let result: ReturnType<typeof useChat> | undefined
4721+
/** Like home.tsx: the empty-state composer swaps for the chat view's once messages show. */
4722+
function HomeLike() {
4723+
result = useChat('ws-1', undefined)
4724+
return result.messages.length > 0 ? (
4725+
<section key='chat'>
4726+
<ModelSelector />
4727+
</section>
4728+
) : (
4729+
<main key='empty'>
4730+
<ModelSelector />
4731+
</main>
4732+
)
4733+
}
4734+
act(() => {
4735+
root.render(
4736+
<QueryClientProvider client={queryClient}>
4737+
<HomeLike />
4738+
</QueryClientProvider>
4739+
)
4740+
})
4741+
const shownEffort = () =>
4742+
container.querySelector('[aria-label="Reasoning effort"]')?.getAttribute('aria-description')
4743+
act(() => useMothershipEffortStore.getState().setNewChatEffort('low'))
4744+
expect(shownEffort()).toBe('Low')
4745+
4746+
await act(async () => {
4747+
void result?.sendMessage('Plan the launch')
4748+
})
4749+
await waitFor(() => state.postBodies.length === 1)
4750+
expect(state.postBodies[0].effort).toBe('low')
4751+
expect(container.querySelector('section')).not.toBeNull()
4752+
expect(shownEffort()).toBe('Low')
4753+
4754+
await act(async () => {
4755+
post.reject(new TypeError('Failed to fetch'))
4756+
})
4757+
await waitFor(() => container.querySelector('main') !== null)
4758+
4759+
expect(useMothershipEffortStore.getState().newChatEffort).toBe('low')
4760+
expect(shownEffort()).toBe('Low')
4761+
})
4762+
4763+
it.each(['leaves the page', 'opens another chat'] as const)(
4764+
'drops an unsent new-chat effort when the surface %s',
4765+
(leave) => {
4766+
useMothershipEffortStore.getState().reset()
4767+
;(globalThis as { IS_REACT_ACT_ENVIRONMENT?: boolean }).IS_REACT_ACT_ENVIRONMENT = true
4768+
queryClient = new QueryClient({ defaultOptions: { queries: { retry: false } } })
4769+
const root = createRoot(document.createElement('div'))
4770+
mountedRoots.push(root)
4771+
function Surface({ chatId }: { chatId?: string }) {
4772+
useChat('ws-1', chatId)
4773+
return null
4774+
}
4775+
const render = (chatId?: string) =>
4776+
act(() =>
4777+
root.render(
4778+
<QueryClientProvider client={queryClient}>
4779+
<Surface chatId={chatId} />
4780+
</QueryClientProvider>
4781+
)
4782+
)
4783+
render()
4784+
useMothershipEffortStore.getState().setNewChatEffort('low')
4785+
4786+
if (leave === 'leaves the page') act(() => root.unmount())
4787+
else render('chat-other')
4788+
4789+
expect(useMothershipEffortStore.getState().newChatEffort).toBeNull()
4790+
}
4791+
)
4792+
47044793
it('loads the saved transcript once when its own stream completes', async () => {
47054794
const chatId = 'chat-own-completion'
47064795
const history: MothershipChatHistory = {

‎apps/sim/app/workspace/[workspaceId]/home/hooks/use-chat.ts‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1009,6 +1009,13 @@ export function useChat(
10091009
surfaceMountedRef.current = false
10101010
}
10111011
}, [])
1012+
/* The new-chat effort pick belongs to this surface, not to one composer: it outlives the swap
1013+
from the empty-state composer to the chat view during a first send, and drops only when the
1014+
surface leaves the new chat. */
1015+
useEffect(() => {
1016+
if (initialChatId) return
1017+
return () => useMothershipEffortStore.getState().setNewChatEffort(null)
1018+
}, [initialChatId])
10121019
const tableViewContextsRef = useRef({
10131020
scopeId: desktopScopeId,
10141021
views: new Map<string, MothershipTableViewContext>(),

‎apps/sim/hooks/queries/mothership-chats.test.ts‎

Lines changed: 22 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -11,7 +11,7 @@ const { suspendBrowserScope, suspendTerminalScope, clearChat } = vi.hoisted(() =
1111
}))
1212

1313
vi.mock('@/stores/mothership-queue/store', () => ({
14-
useMothershipQueueStore: { getState: () => ({ clearChat }) },
14+
useMothershipQueueStore: { getState: () => ({ clearChat, cleared: {} }) },
1515
}))
1616

1717
vi.mock('@tanstack/react-query', () => reactQueryMock)
@@ -26,6 +26,7 @@ vi.mock('@/lib/terminal/transport', () => ({
2626

2727
import type { MothershipEffort } from '@/lib/mothership/model-options'
2828
import {
29+
fetchMothershipChatHistory,
2930
useDeleteMothershipChats,
3031
useSetMothershipChatEffort,
3132
} from '@/hooks/queries/mothership-chats'
@@ -100,6 +101,26 @@ describe('tasks query boundary parsing', () => {
100101
})
101102
})
102103

104+
it('loads a chat from a server that predates the effort field', async () => {
105+
vi.mocked(fetch).mockResolvedValueOnce(
106+
jsonResponse({
107+
success: true,
108+
chat: {
109+
id: 'chat-1',
110+
title: null,
111+
mode: 'agent',
112+
messages: [],
113+
activeStreamId: null,
114+
resources: [],
115+
},
116+
})
117+
)
118+
119+
const history = await fetchMothershipChatHistory('chat-1')
120+
121+
expect(history.effort).toBeNull()
122+
})
123+
103124
it('keeps the latest effort pick when an earlier queued save of the same value fails', async () => {
104125
const tanstack =
105126
await vi.importActual<typeof import('@tanstack/react-query')>('@tanstack/react-query')

‎apps/sim/lib/api/contracts/mothership-chats.ts‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -420,7 +420,8 @@ export const getMothershipChatResponseSchema = z.object({
420420
messages: z.array(z.unknown()),
421421
activeStreamId: z.string().nullable(),
422422
resources: z.array(z.unknown()),
423-
effort: mothershipChatEffortChoiceSchema,
423+
/** Optional so a client still loads chats from a server that predates the field. */
424+
effort: mothershipChatEffortChoiceSchema.optional(),
424425
createdAt: z.union([z.string(), z.date()]).nullable().optional(),
425426
updatedAt: z.union([z.string(), z.date()]).nullable().optional(),
426427
streamSnapshot: mothershipChatStreamSnapshotSchema.optional(),

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

Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -133,6 +133,22 @@ describe('createWorkspaceSSE', () => {
133133
expect(unsubscribe).toHaveBeenCalledTimes(1)
134134
})
135135

136+
it('never subscribes when the request aborted before the stream started', async () => {
137+
const controller = new AbortController()
138+
controller.abort()
139+
const subscribe = vi.fn(() => () => {})
140+
const { body } = await openConnection(controller.signal, [{ subscribe }])
141+
let closed = false
142+
void drain(body).then(() => {
143+
closed = true
144+
})
145+
146+
await vi.advanceTimersByTimeAsync(0)
147+
148+
expect(closed).toBe(true)
149+
expect(subscribe).not.toHaveBeenCalled()
150+
})
151+
136152
it('runs every teardown when one unsubscribe throws', async () => {
137153
const first = vi.fn(() => {
138154
throw new Error('unsubscribe failed')

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

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -213,6 +213,13 @@ export function createSSEStream(request: NextRequest, config: SSEStreamConfig):
213213
)
214214
}
215215

216+
// An abort listener never fires for a signal that is already aborted, so a client that left
217+
// while the route was authorizing would otherwise hold its subscriptions until rotation.
218+
if (request.signal.aborted) {
219+
close('aborted')
220+
return
221+
}
222+
216223
try {
217224
// The runtime sends the status and headers with the first body chunk, so the stream writes
218225
// one as soon as it opens. A client reads its state once the stream opens, so it opens once

‎apps/sim/stores/mothership-effort/store.ts‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -15,8 +15,8 @@ interface MothershipEffortState {
1515
setModel: (model: ModelSelection['model']) => void
1616
setFastMode: (fastMode: boolean) => void
1717
/**
18-
* The effort picked in a composer whose chat does not exist yet. Its first send records
19-
* it on the new chat; leaving that composer unsent drops it.
18+
* The effort picked on a chat surface whose chat does not exist yet. Its first send records
19+
* it on the new chat; leaving that surface unsent drops it.
2020
*/
2121
newChatEffort: MothershipEffort | null
2222
setNewChatEffort: (effort: MothershipEffort | null) => void

0 commit comments

Comments
 (0)