fix(chat): detach only admitted turns and report stale terminal calls

Author: waleedlatif1

apps/sim/app/workspace/[workspaceId]/home/hooks/stream-protocol.ts

@@ -22,6 +22,7 @@ import type { StreamBatchEvent } from '@/lib/mothership/request/session/types'
 
 /** Both live transports heartbeat every 15s; three missed heartbeats trigger cursor recovery. */
 export const STREAM_IDLE_TIMEOUT_MS = 45_000
+export const STREAM_BATCH_FETCH_TIMEOUT_MS = 10_000
 
 export type StreamBatchResponse = {
   success: boolean

apps/sim/app/workspace/[workspaceId]/home/hooks/stream/client-tool-start.ts

@@ -29,17 +29,12 @@ export type ClientToolStart = ClientToolCall &
   )
 
 /**
- * Resolves the client-executed tool call a stream event asks this client to
- * start, or null when the event starts nothing. Only a complete call frame
- * that is not held behind an approval prompt starts a tool.
- *
- * @param isPending - whether the call is still running without a result, as
- *   far as the caller has seen
+ * Resolves the client-executed tool call a stream event hands this client, or
+ * null when the event starts nothing. Only a complete call frame that is not
+ * held behind an approval prompt starts a tool; whether the call is still
+ * pending is the caller's to decide from what it has seen.
  */
-export function resolveClientToolStart(
-  event: ToolEvent,
-  isPending: (toolCallId: string) => boolean
-): ClientToolStart | null {
+export function resolveClientToolStart(event: ToolEvent): ClientToolStart | null {
   const payload = event.payload
   if (
     'previewPhase' in payload ||
@@ -55,15 +50,11 @@ export function resolveClientToolStart(
   const { toolCallId, toolName } = payload
   const args = payload.arguments as Record<string, unknown> | undefined
   const call = { toolCallId, args: args ?? {}, eventTs: event.ts }
-  let start: ClientToolStart | null = null
-  if (isWorkflowToolName(toolName)) {
-    start = { ...call, kind: 'workflow', toolName }
-  } else if (isNativeFileTool(toolName) || isUserLocalVfsToolCall(toolName, args)) {
-    start = { ...call, kind: 'localFilesystem', toolName }
-  } else if (isCurrentBrowserToolName(toolName)) {
-    start = { ...call, kind: 'browser', toolName }
-  } else if (isTerminalToolName(toolName)) {
-    start = { ...call, kind: 'terminal', toolName }
+  if (isWorkflowToolName(toolName)) return { ...call, kind: 'workflow', toolName }
+  if (isNativeFileTool(toolName) || isUserLocalVfsToolCall(toolName, args)) {
+    return { ...call, kind: 'localFilesystem', toolName }
   }
-  return start && isPending(toolCallId) ? start : null
+  if (isCurrentBrowserToolName(toolName)) return { ...call, kind: 'browser', toolName }
+  if (isTerminalToolName(toolName)) return { ...call, kind: 'terminal', toolName }
+  return null
 }

apps/sim/app/workspace/[workspaceId]/home/hooks/stream/detached-client-tools.ts

@@ -40,11 +40,20 @@ import {
   parseStreamBatchResponse,
   resolveChatIdFromStreamBatch,
   resolveChatIdFromStreamEvent,
+  STREAM_BATCH_FETCH_TIMEOUT_MS,
   STREAM_IDLE_TIMEOUT_MS,
 } from '@/app/workspace/[workspaceId]/home/hooks/stream-protocol'
 
 const logger = createLogger('DetachedClientTools')
 
+/**
+ * Responses after which there is nothing left to relay: no run exists for the
+ * stream (404), or this client can no longer read it (401, 403).
+ */
+function isRelayEndStatus(status: number): boolean {
+  return status === 404 || status === 401 || status === 403
+}
+
 /** A live turn the user left. */
 export interface DetachedChatTurn {
   streamId: string
@@ -60,6 +69,7 @@ export interface DetachedChatTurn {
 
 interface Relay {
   controller: AbortController
+  /** Known up front or read off the stream; tools run in this chat's desktop scope. */
   chatId?: string
 }
 
@@ -132,8 +142,8 @@ async function relayClientTools(turn: DetachedChatTurn, relay: Relay): Promise<v
       handledToolCallIds.add(settledId)
       return
     }
-    const start = resolveClientToolStart(event, (id) => !handledToolCallIds.has(id))
-    if (!start) return
+    const start = resolveClientToolStart(event)
+    if (!start || handledToolCallIds.has(start.toolCallId)) return
     handledToolCallIds.add(start.toolCallId)
     if (!relay.chatId) {
       logger.error('Detached client tool arrived before its chat id', {
@@ -145,14 +155,14 @@ async function relayClientTools(turn: DetachedChatTurn, relay: Relay): Promise<v
     startDetachedClientTool(turn, relay.chatId, start)
   }
 
-  /** Reads the events past the cursor at once; resolves true once the run is over. */
+  /** Reads the events past the cursor at once; resolves true once there is nothing left to relay. */
   const readBatch = async (): Promise<boolean> => {
     // boundary-raw-fetch: stream-resume batch endpoint needs per-request traceparent propagation the contract layer does not model
     const response = await fetch(buildStreamResumeUrl(streamId, cursor, { batch: true }), {
-      signal,
+      signal: AbortSignal.any([signal, AbortSignal.timeout(STREAM_BATCH_FETCH_TIMEOUT_MS)]),
       headers,
     })
-    if (response.status === 404) return true
+    if (isRelayEndStatus(response.status)) return true
     if (!response.ok) throw new Error(`Stream batch responded with status ${response.status}`)
     const batch = parseStreamBatchResponse(await response.json())
     relay.chatId ??= resolveChatIdFromStreamBatch(batch)
@@ -164,11 +174,11 @@ async function relayClientTools(turn: DetachedChatTurn, relay: Relay): Promise<v
     return isTerminalStreamStatus(batch.status)
   }
 
-  /** Tails one live connection; resolves true once the run is over. */
+  /** Tails one live connection; resolves true once there is nothing left to relay. */
   const readTail = async (): Promise<boolean> => {
     // boundary-raw-fetch: live SSE tail endpoint streams events consumed via readSSELines
     const response = await fetch(buildStreamResumeUrl(streamId, cursor), { signal, headers })
-    if (response.status === 404) return true
+    if (isRelayEndStatus(response.status)) return true
     if (!response.ok || !response.body) {
       throw new Error(`Stream tail responded with status ${response.status}`)
     }
@@ -224,13 +234,10 @@ export function detachClientTools(turn: DetachedChatTurn): void {
 }
 
 /**
- * Stops relaying a chat whose view reads its stream again. Tools the relay
- * already started keep running and report their outcome.
+ * Stops relaying a stream the chat view reads again. Tools the relay already
+ * started keep running and report their outcome.
  */
-export function reattachClientTools(chatId: string): void {
-  for (const [streamId, relay] of relays) {
-    if (relay.chatId !== chatId) continue
-    relay.controller.abort('chat_reattached')
-    relays.delete(streamId)
-  }
+export function reattachClientTools(streamId: string): void {
+  relays.get(streamId)?.controller.abort('chat_reattached')
+  relays.delete(streamId)
 }

apps/sim/app/workspace/[workspaceId]/home/hooks/stream/handle-tool-event.ts

@@ -202,11 +202,12 @@ export function handleToolEvent(ctx: StreamLoopContext, parsed: ToolEvent): void
   // reducer, run its side effects now (the result event had no node to act on).
   if (node?.kind === 'tool' && node.result) runToolResultSideEffects(ctx, node, replay)
 
-  const start = resolveClientToolStart(parsed, (toolCallId) => {
-    if (deps.options.suppressedWorkflowToolStartIds?.has(toolCallId)) return false
-    const tool = state.model.nodes.get(resolveToolId(state.model, toolCallId))
-    return tool?.kind === 'tool' && tool.status === 'running' && !tool.result
-  })
-  if (start) startClientTool(deps, start)
+  const start = resolveClientToolStart(parsed)
+  const isPending =
+    node?.kind === 'tool' &&
+    node.status === 'running' &&
+    !node.result &&
+    !deps.options.suppressedWorkflowToolStartIds?.has(rawId)
+  if (start && isPending) startClientTool(deps, start)
   ops.flush()
 }

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

@@ -178,6 +178,7 @@ import {
   isStreamSchemaValidationError,
   parseStreamBatchResponse,
   resolveChatIdFromStreamBatch,
+  STREAM_BATCH_FETCH_TIMEOUT_MS,
   STREAM_IDLE_TIMEOUT_MS,
   type StreamBatchResponse,
   StreamGoneError,
@@ -293,7 +294,6 @@ const MAX_RECONNECT_ATTEMPTS = 10
 const RECONNECT_BASE_DELAY_MS = 1000
 const RECONNECT_MAX_DELAY_MS = 30_000
 const RECONNECT_EXHAUSTED_RECHECK_MS = 30_000
-const STREAM_BATCH_FETCH_TIMEOUT_MS = 10_000
 const STREAM_CHAT_ID_RESOLVE_TIMEOUT_MS = 10_000
 const CHAT_HISTORY_RECOVERY_TIMEOUT_MS = 10_000
 const STOP_REQUEST_TIMEOUT_MS = 15_000
@@ -1609,23 +1609,22 @@ export function useChat(
 
   /**
    * Hands the live turn's client tools to a detached relay as the user leaves
-   * its chat, so a desktop run keeps going in the background. A turn is live
-   * once the server admitted it: its stream id is known, or its response is
-   * being read (a send's stream id is its user message id). When a relay takes
-   * the turn, the caller releases the turn's controller without aborting it,
-   * so tools already running report their outcome instead of being cancelled.
+   * its chat, so a desktop run keeps going in the background. Only a turn the
+   * server admitted is detached; a send still awaiting admission is withdrawn
+   * by the unmount cleanup and handed to the next chat surface instead. When a
+   * relay takes the turn, the caller releases the turn's controller without
+   * aborting it, so tools already running report their outcome.
    *
    * @returns whether a relay took the turn
    */
   detachLiveTurnClientToolsRef.current = (): boolean => {
-    const streamId =
-      streamIdRef.current ??
-      (streamReaderRef.current ? activeTurnRef.current?.userMessageId : undefined)
+    const streamId = streamIdRef.current
     if (
       !isDesktopApp() ||
       requestModeRef.current === 'assistant' ||
       !sendingRef.current ||
-      !streamId
+      !streamId ||
+      pendingChatAdmissionRef.current
     ) {
       return false
     }
@@ -2210,8 +2209,7 @@ export function useChat(
         return { sawStreamError: false, sawComplete: false }
       }
       streamReaderRef.current = reader
-      const readerChatId = options?.targetChatId ?? chatIdRef.current
-      if (readerChatId) reattachClientTools(readerChatId)
+      if (streamIdRef.current) reattachClientTools(streamIdRef.current)
 
       try {
         await readSSELines(reader, {
@@ -2563,6 +2561,7 @@ export function useChat(
           }
 
           if (isStaleReconnect()) {
+            await sseRes.body.cancel()
             return { error: false, aborted: true }
           }
 

apps/sim/lib/mothership/tools/client/launch-local-filesystem-tool.ts

@@ -1,18 +1,9 @@
 import { createLogger } from '@sim/logger'
-import { LRUCache } from 'lru-cache'
 import type { LocalFilesystemExecutionContext } from '@/lib/mothership/tools/client/local-filesystem'
 import { isNativeFileTool, isUserLocalVfsToolCall } from '@/lib/mothership/tools/local-filesystem'
 
 const logger = createLogger('CopilotLocalFilesystemTool')
 
-/**
- * Exactly-once guard. A call's chat view and the relay that runs it after the
- * user leaves can both see it, and a remounted view replays calls still
- * running, so the guard lives here rather than with any one caller. Bounded:
- * a call evicted behind this many newer ones is long settled.
- */
-const launchedToolCallIds = new LRUCache<string, true>({ max: 500 })
-
 /**
  * Runs a local filesystem tool call on the desktop client. The executor is
  * loaded on demand: it only runs for desktop-local calls, and a static import
@@ -33,8 +24,6 @@ export function launchLocalFilesystemTool(
   ) {
     return
   }
-  if (launchedToolCallIds.has(toolCallId)) return
-  launchedToolCallIds.set(toolCallId, true)
 
   import('@/lib/mothership/tools/client/local-filesystem').then(
     (m) => m.executeLocalFilesystemTool(toolCallId, toolName, args, context),

apps/sim/lib/mothership/tools/client/local-filesystem.ts

@@ -14,6 +14,7 @@ import {
 } from '@sim/desktop-bridge/local-filesystem-limits'
 import { createLogger } from '@sim/logger'
 import { toError } from '@sim/utils/errors'
+import { LRUCache } from 'lru-cache'
 import micromatch from 'micromatch'
 import { getDesktopBridge } from '@/lib/desktop'
 import { ASYNC_TOOL_CONFIRMATION_STATUS } from '@/lib/mothership/async-runs/lifecycle'
@@ -337,12 +338,21 @@ async function execute(
   return executeUserLocalRead(toolCallId, args, context.signal)
 }
 
+/**
+ * Exactly-once guard. A call's chat view and the relay that runs it after the
+ * user leaves can both start it, and a remounted view replays calls still
+ * running. Bounded: a call evicted behind this many newer ones is long settled.
+ */
+const executedToolCallIds = new LRUCache<string, true>({ max: 500 })
+
 export function executeLocalFilesystemTool(
   toolCallId: string,
   toolName: string,
   args: Record<string, unknown>,
   context: LocalFilesystemExecutionContext
 ): void {
+  if (executedToolCallIds.has(toolCallId)) return
+  executedToolCallIds.set(toolCallId, true)
   if (isNativeFileTool(toolName)) {
     void executeNativeFileTool(toolCallId, toolName, context.signal)
     return

apps/sim/lib/mothership/tools/client/terminal-tool-execution.ts

@@ -25,6 +25,9 @@ const logger = createLogger('CopilotTerminalToolExecution')
 
 /** Tool events older than this are replays, not live instructions. */
 const MAX_EVENT_AGE_MS = 120_000
+/** A stale call never ran in this client; the waiter must hear so or the turn hangs. */
+const STALE_EVENT_MESSAGE =
+  'This terminal command was delivered too late to run safely, so it was not run. Ask again to retry it.'
 const EXECUTED_STORAGE_PREFIX = 'sim:copilot:terminal-tool-executed:'
 
 /**
@@ -121,6 +124,17 @@ export function executeTerminalToolOnClient(
   const age = eventAgeMs(eventTs)
   if (age !== null && age > MAX_EVENT_AGE_MS) {
     logger.info('Skipping stale terminal tool event', { toolCallId, operation, age })
+    void reportClientToolCompletion(
+      toolCallId,
+      ASYNC_TOOL_CONFIRMATION_STATUS.error,
+      STALE_EVENT_MESSAGE,
+      { error: STALE_EVENT_MESSAGE, staleEvent: true }
+    ).catch((reportErr) => {
+      logger.error('Failed to report stale terminal tool event', {
+        toolCallId,
+        error: toError(reportErr).message,
+      })
+    })
     return
   }
   markExecuted(toolCallId)