fix(chat): never re-send on an unresolved probe, and reconnect after adopting

Two defects in the orphaned-stream probe, both found by Bugbot.

A probe cut short by an epoch change (unmount, chat switch) returned the same
`undefined` as "the server has no such stream", so the dispatcher fell through
to `startSendMessage`. After unmount the teardown has already dropped the abort
controller, so that send opened a POST nothing could cancel — duplicating the
very message this recovery exists to protect. The probe now reports
`superseded` distinctly and the dispatcher leaves the entry queued, keeping its
`recoverStreamId` so a later mount probes again.

Adopting the recovered chat also invalidated only the chat list. Hydration
reconnects to a live turn solely on `chatHistory.activeStreamId`, and that
query is cached for MOTHERSHIP_CHAT_HISTORY_STALE_TIME — on a chat-bound
recover the client normally holds a copy predating this stream, so the adopted
chat rendered with the running response invisible. Adoption now invalidates the
chat detail too.

Both regression tests were confirmed to fail without their fix: the first
re-sends (2 POSTs instead of 1), the second never invalidates. The probe stub
gained a `pending` mode because a `gone` probe answers on the first attempt and
leaves nothing in flight to interrupt — the earlier draft of the first test
passed with the guard removed and proved nothing.

Author: waleedlatif1

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

@@ -47,18 +47,22 @@ interface NetworkState {
   postBehavior: 'hang' | 'accept'
   postCalls: number
   /**
-   * Chat the orphaned-stream probe resolves to, standing in for a request the
-   * server accepted before the client's cleanup abort tore the socket down.
-   * `null` means the server has no such stream (it never accepted the request).
+   * How the orphaned-stream probe answers:
+   * - `found` — the server accepted the withdrawn request and owns a chat
+   * - `gone` — 404, it has no such stream (never accepted)
+   * - `pending` — registered but no owner yet, so the probe keeps polling;
+   *   this is the only mode that leaves a probe in flight to interrupt
    */
-  orphanedStreamChatId: string | null
+  probeBehavior: 'found' | 'gone' | 'pending'
+  orphanedStreamChatId: string
   streamProbes: number
 }
 
 const state: NetworkState = {
   postBehavior: 'hang',
   postCalls: 0,
-  orphanedStreamChatId: null,
+  probeBehavior: 'gone',
+  orphanedStreamChatId: 'chat-server-already-made',
   streamProbes: 0,
 }
 
@@ -81,15 +85,16 @@ async function fetchStub(input: RequestInfo | URL, init?: RequestInit): Promise<
     state.streamProbes++
     // 404 is what the server returns for a stream it never registered — i.e.
     // the request really was withdrawn before it was accepted.
-    if (!state.orphanedStreamChatId) {
+    if (state.probeBehavior === 'gone') {
       return new Response(JSON.stringify({ error: 'stream gone' }), { status: 404 })
     }
     return new Response(
       JSON.stringify({
         success: true,
         events: [],
         status: 'streaming',
-        chatId: state.orphanedStreamChatId,
+        // `pending` omits the owner, so the probe keeps polling.
+        ...(state.probeBehavior === 'found' ? { chatId: state.orphanedStreamChatId } : {}),
       }),
       { status: 200, headers: { 'Content-Type': 'application/json' } }
     )
@@ -205,7 +210,7 @@ describe('useChat remount send recovery', () => {
     vi.stubGlobal('fetch', fetchStub)
     state.postBehavior = 'hang'
     state.postCalls = 0
-    state.orphanedStreamChatId = null
+    state.probeBehavior = 'gone'
     state.streamProbes = 0
     mockRequestJson.mockResolvedValue({ chats: [] })
     useMothershipQueueStore.setState({ queues: {}, editing: {} })
@@ -334,7 +339,7 @@ describe('useChat remount send recovery', () => {
   describe('recovered send probes the orphaned stream before re-sending', () => {
     it('adopts the chat the server already created instead of sending twice', async () => {
       // The server accepted the withdrawn request and registered its stream.
-      state.orphanedStreamChatId = 'chat-server-already-made'
+      state.probeBehavior = 'found'
 
       const { getResult, unmount } = renderUseChat()
       await act(async () => {
@@ -361,9 +366,85 @@ describe('useChat remount send recovery', () => {
       expect(allQueuedMessages()).toHaveLength(0)
     })
 
+    /**
+     * Adoption alone does not surface the running turn: hydration reconnects
+     * only when `chatHistory.activeStreamId` is set, and that query is cached
+     * for 30s. Without an explicit detail invalidation the adopted chat renders
+     * with the live response invisible.
+     */
+    it('invalidates the adopted chat detail so hydration can reconnect', async () => {
+      state.probeBehavior = 'found'
+
+      const { getResult, unmount } = renderUseChat()
+      await act(async () => {
+        void getResult().sendMessage('surface the running turn')
+      })
+      await waitFor(() => state.postCalls === 1)
+      unmount()
+      await waitFor(() => window.localStorage.getItem('sim_mothership_handoff') !== null)
+
+      const handoff = MothershipHandoffStorage.consume('ws-1')
+      const replacement = renderUseChat()
+      const invalidateSpy = vi.spyOn(queryClient, 'invalidateQueries')
+
+      await act(async () => {
+        void replacement.getResult().sendMessage(handoff?.message as string, undefined, undefined, {
+          recoverStreamId: handoff?.recoverStreamId as string,
+        })
+      })
+      await waitFor(() =>
+        invalidateSpy.mock.calls.some(([arg]) => {
+          const key = (arg as { queryKey?: unknown[] } | undefined)?.queryKey
+          return Array.isArray(key) && key.includes('chat-server-already-made')
+        })
+      )
+
+      expect(state.postCalls).toBe(1)
+    })
+
+    /**
+     * A probe cut short by unmount answers "unknown", not "safe to send".
+     * Falling through to `startSendMessage` there would open a POST whose
+     * abort controller the teardown already dropped — an uncancellable request
+     * that duplicates the send. The entry must stay queued instead.
+     */
+    it('does not send when the probe is cut short by unmount', async () => {
+      state.probeBehavior = 'gone'
+      const { getResult, unmount } = renderUseChat()
+      await act(async () => {
+        void getResult().sendMessage('do not zombie me')
+      })
+      await waitFor(() => state.postCalls === 1)
+      unmount()
+      await waitFor(() => window.localStorage.getItem('sim_mothership_handoff') !== null)
+
+      const handoff = MothershipHandoffStorage.consume('ws-1')
+
+      /* `pending` keeps the probe polling instead of answering on the first
+         attempt, which is what leaves one in flight to interrupt. A `gone`
+         probe answers immediately and the re-send would already have happened
+         before the unmount — that version of this test cannot fail. */
+      state.probeBehavior = 'pending'
+      const replacement = renderUseChat()
+      await act(async () => {
+        void replacement.getResult().sendMessage(handoff?.message as string, undefined, undefined, {
+          recoverStreamId: handoff?.recoverStreamId as string,
+        })
+      })
+      await waitFor(() => state.streamProbes > 0)
+      const postsBeforeUnmount = state.postCalls
+      replacement.unmount()
+
+      // Well past the probe's poll budget: nothing may send after teardown.
+      await act(async () => {
+        await sleep(3000)
+      })
+      expect(state.postCalls).toBe(postsBeforeUnmount)
+    })
+
     it('re-sends when the server has no stream for it', async () => {
       // 404 from the probe: the request really was withdrawn before acceptance.
-      state.orphanedStreamChatId = null
+      state.probeBehavior = 'gone'
 
       const { getResult, unmount } = renderUseChat()
       await act(async () => {

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

@@ -153,6 +153,17 @@ export interface SendMessageOptions {
  */
 type StartSendMessageResult = boolean | { kind: 'recoverable_cleanup_abort'; streamId: string }
 
+/**
+ * Outcome of asking the server whether it accepted the request a cleanup abort
+ * withdrew. `superseded` is deliberately distinct from `not_found`: the former
+ * means the answer is unknown because this dispatch is no longer current, and
+ * must not be treated as licence to re-send.
+ */
+type RecoveredSendProbe =
+  | { status: 'adopted'; chatId: string }
+  | { status: 'not_found' }
+  | { status: 'superseded' }
+
 export interface UseChatReturn {
   messages: ChatMessage[]
   isSending: boolean
@@ -4513,22 +4524,28 @@ export function useChat(
    * the caller re-sends.
    */
   const resolveRecoveredSendChatId = useCallback(
-    async (streamId: string, epoch: number): Promise<string | undefined> => {
+    async (streamId: string, epoch: number): Promise<RecoveredSendProbe> => {
       const deadline = Date.now() + RECOVERED_SEND_PROBE_TIMEOUT_MS
       while (true) {
         const resolve = resolveDetachedChatForStreamRef.current
-        if (!resolve) return undefined
-        // Stop as soon as this dispatch is superseded (chat switch, unmount).
-        // The caller adopts what this returns, which rewrites the URL, and
-        // doing that after the user moved on would hijack their navigation.
-        if (epoch !== queueDispatchEpochRef.current) return undefined
+        if (!resolve) return { status: 'not_found' }
+        /* "Superseded" is NOT the same answer as "the server has no such
+           stream", and the caller must not conflate them: adopting rewrites
+           the URL, and re-sending after an unmount would open a POST whose
+           abort controller the teardown has already dropped — a zombie request
+           nothing can cancel, duplicating the very send this recovery exists
+           to protect. Reported distinctly so the caller leaves the entry
+           queued for a later mount instead. */
+        if (epoch !== queueDispatchEpochRef.current) return { status: 'superseded' }
         const resolution = await resolve(streamId)
-        if (epoch !== queueDispatchEpochRef.current) return undefined
-        if (resolution.chatId) return resolution.chatId
+        if (epoch !== queueDispatchEpochRef.current) return { status: 'superseded' }
+        if (resolution.chatId) return { status: 'adopted', chatId: resolution.chatId }
         // A terminal status means the stream existed and finished without a
         // durable owner; polling cannot improve on that.
-        if (resolution.terminal) return undefined
-        if (Date.now() + RECOVERED_SEND_PROBE_INTERVAL_MS >= deadline) return undefined
+        if (resolution.terminal) return { status: 'not_found' }
+        if (Date.now() + RECOVERED_SEND_PROBE_INTERVAL_MS >= deadline) {
+          return { status: 'not_found' }
+        }
         await sleep(RECOVERED_SEND_PROBE_INTERVAL_MS)
       }
     },
@@ -4640,16 +4657,23 @@ export function useChat(
            the chat and the billed run. Probe the orphaned stream first and
            adopt its chat instead when it exists. */
         if (liveMsg.recoverStreamId) {
-          const adoptedChatId = await resolveRecoveredSendChatId(
-            liveMsg.recoverStreamId,
-            options.epoch
-          )
-          if (adoptedChatId) {
+          const probe = await resolveRecoveredSendChatId(liveMsg.recoverStreamId, options.epoch)
+          /* Unknown, not "safe to send" — leave the entry queued (it keeps its
+             `recoverStreamId`) so the next mount's drain probes again. */
+          if (probe.status === 'superseded') return
+          if (probe.status === 'adopted') {
             removeQueuedMessage()
-            adoptResolvedChatId(adoptedChatId, {
+            adoptResolvedChatId(probe.chatId, {
               replaceHomeHistory: true,
               invalidateList: true,
             })
+            /* Adoption alone does not surface the running turn. Hydration only
+               reconnects when `chatHistory.activeStreamId` is set, and that
+               query is cached for `MOTHERSHIP_CHAT_HISTORY_STALE_TIME` — on a
+               chat-bound recover the client usually holds a copy predating this
+               stream, so without an explicit detail invalidation the adopted
+               chat renders with the live response invisible. */
+            invalidateChatQueries({ includeDetail: true, targetChatId: probe.chatId })
             return
           }
         }
@@ -4677,7 +4701,13 @@ export function useChat(
         userRemovedDuringDispatchRef.current.delete(msg.id)
       }
     },
-    [startSendMessage, workspaceId, resolveRecoveredSendChatId, adoptResolvedChatId]
+    [
+      startSendMessage,
+      workspaceId,
+      resolveRecoveredSendChatId,
+      adoptResolvedChatId,
+      invalidateChatQueries,
+    ]
   )
 
   const runQueueDispatchLoop = useCallback(async () => {