fix(chat): keep desktop tools running in chats left mid-turn

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

@@ -11,6 +11,7 @@ import {
 } from '@/lib/mothership/generated/mothership-stream-v1'
 import {
   type ParseStreamEventEnvelopeFailure,
+  type PersistedStreamEventEnvelope,
   parsePersistedStreamEventEnvelope,
 } from '@/lib/mothership/request/session/contract'
 import {
@@ -19,6 +20,10 @@ import {
 } from '@/lib/mothership/request/session/file-preview-session-contract'
 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
   events: StreamBatchEvent[]
@@ -111,18 +116,27 @@ export function parseStreamBatchResponse(value: unknown): StreamBatchResponse {
   }
 }
 
+/** The chat an event names, from its stream metadata or a chat session event. */
+export function resolveChatIdFromStreamEvent(
+  event: PersistedStreamEventEnvelope
+): string | undefined {
+  const streamChatId = typeof event.stream?.chatId === 'string' ? event.stream.chatId : undefined
+  if (streamChatId) return streamChatId
+  if (
+    event.type === MothershipStreamV1EventType.session &&
+    event.payload.kind === MothershipStreamV1SessionKind.chat
+  ) {
+    return event.payload.chatId
+  }
+  return undefined
+}
+
 export function resolveChatIdFromStreamBatch(batch: StreamBatchResponse): string | undefined {
   if (batch.chatId) return batch.chatId
 
   for (const { event } of batch.events) {
-    const streamChatId = typeof event.stream?.chatId === 'string' ? event.stream.chatId : undefined
-    if (streamChatId) return streamChatId
-    if (
-      event.type === MothershipStreamV1EventType.session &&
-      event.payload.kind === MothershipStreamV1SessionKind.chat
-    ) {
-      return event.payload.chatId
-    }
+    const chatId = resolveChatIdFromStreamEvent(event)
+    if (chatId) return chatId
   }
 
   return undefined
@@ -148,6 +162,24 @@ export function isZeroStreamCursor(cursor: string): boolean {
   return Number.isFinite(sequence) && sequence <= 0
 }
 
+/**
+ * The resume endpoint for a stream's events after `afterCursor`: replayed then
+ * tailed live, or returned as one JSON batch.
+ */
+export function buildStreamResumeUrl(
+  streamId: string,
+  afterCursor: string,
+  options?: { batch?: boolean }
+): string {
+  const url = `/api/mothership/chat/stream?streamId=${encodeURIComponent(streamId)}&after=${encodeURIComponent(afterCursor)}`
+  return options?.batch ? `${url}&batch=true` : url
+}
+
+/** The cursor an event advances its stream to; dedupes replayed events. */
+export function getStreamEventCursor(event: PersistedStreamEventEnvelope): string {
+  return event.stream?.cursor ?? String(event.seq)
+}
+
 /**
  * The resume endpoint 404s when no run exists for the stream — there is
  * nothing left to resume, so reconnect falls back to the persisted DB

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

@@ -0,0 +1,60 @@
+import { type CurrentBrowserToolName, isCurrentBrowserToolName } from '@sim/browser-protocol'
+import { isTerminalToolName } from '@sim/terminal-protocol'
+import {
+  MothershipStreamV1ToolPhase,
+  MothershipStreamV1ToolStatus,
+} from '@/lib/mothership/generated/mothership-stream-v1'
+import type { PersistedStreamEventEnvelope } from '@/lib/mothership/request/session/contract'
+import { isNativeFileTool, isUserLocalVfsToolCall } from '@/lib/mothership/tools/local-filesystem'
+import { isWorkflowToolName } from '@/lib/mothership/tools/workflow-tools'
+
+export type ToolEvent = Extract<PersistedStreamEventEnvelope, { type: 'tool' }>
+
+interface ClientToolCall {
+  toolCallId: string
+  args: Record<string, unknown>
+  /** The envelope's emission timestamp; executors drop stale replays by it. */
+  eventTs?: string
+}
+
+/**
+ * A tool call the orchestrator hands to this client, by the executor that runs
+ * it. The orchestrator blocks until the client reports the call's outcome, so
+ * every start must reach its executor.
+ */
+export type ClientToolStart = ClientToolCall &
+  (
+    | { kind: 'workflow' | 'localFilesystem' | 'terminal'; toolName: string }
+    | { kind: 'browser'; toolName: CurrentBrowserToolName }
+  )
+
+/**
+ * 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): ClientToolStart | null {
+  const payload = event.payload
+  if (
+    'previewPhase' in payload ||
+    payload.phase === MothershipStreamV1ToolPhase.args_delta ||
+    payload.phase === MothershipStreamV1ToolPhase.result ||
+    payload.partial === true ||
+    payload.status === MothershipStreamV1ToolStatus.generating ||
+    payload.status === MothershipStreamV1ToolStatus.awaiting_approval
+  ) {
+    return null
+  }
+
+  const { toolCallId, toolName } = payload
+  const args = payload.arguments as Record<string, unknown> | undefined
+  const call = { toolCallId, args: args ?? {}, eventTs: event.ts }
+  if (isWorkflowToolName(toolName)) return { ...call, kind: 'workflow', toolName }
+  if (isNativeFileTool(toolName) || isUserLocalVfsToolCall(toolName, args)) {
+    return { ...call, kind: 'localFilesystem', toolName }
+  }
+  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

@@ -0,0 +1,243 @@
+/**
+ * Keeps a chat's client-executed tools running after the user leaves it
+ * mid-turn. Leaving detaches the chat view but not the server run, and that
+ * run's browser, terminal, and local filesystem tools execute only in this
+ * client: the orchestrator blocks until the client reports each outcome, so
+ * without a reader the left chat stalls at its next such call. A relay reads
+ * the left turn's stream headlessly and starts those tools until the run ends
+ * or the chat's view reads the stream again. Relays live at module scope
+ * because switching chats remounts the chat surface that detached them. Each
+ * holds one resume connection, and only while its run is live.
+ */
+import { createLogger } from '@sim/logger'
+import { getErrorMessage } from '@sim/utils/errors'
+import { interruptibleSleep } from '@sim/utils/helpers'
+import { backoffWithJitter } from '@sim/utils/retry'
+import { readSSELines } from '@/lib/core/utils/sse'
+import { desktopChatScopeId } from '@/lib/desktop/chat-scope'
+import {
+  MothershipStreamV1EventType,
+  MothershipStreamV1ToolPhase,
+} from '@/lib/mothership/generated/mothership-stream-v1'
+import {
+  isTerminalStreamStatus,
+  type PersistedStreamEventEnvelope,
+  parsePersistedStreamEventEnvelopeJson,
+} from '@/lib/mothership/request/session/contract'
+import { executeBrowserToolOnClient } from '@/lib/mothership/tools/client/browser-tool-execution'
+import { launchLocalFilesystemTool } from '@/lib/mothership/tools/client/launch-local-filesystem-tool'
+import { executeTerminalToolOnClient } from '@/lib/mothership/tools/client/terminal-tool-execution'
+import {
+  type ClientToolStart,
+  resolveClientToolStart,
+} from '@/app/workspace/[workspaceId]/home/hooks/stream/client-tool-start'
+import {
+  buildStreamResumeUrl,
+  createStreamSchemaValidationError,
+  getStreamEventCursor,
+  isAlreadyProcessedStreamCursor,
+  isStreamSchemaValidationError,
+  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
+  /** The turn's chat when the view knew it; a new chat's id is read off the stream. */
+  chatId?: string
+  /** The last cursor the chat view dispatched; the relay resumes after it. */
+  afterCursor: string
+  traceparent?: string
+  workspaceId?: string
+  /** The owner key the chat's desktop scope derives from. */
+  scopeKey: string
+}
+
+interface Relay {
+  controller: AbortController
+  /** Known up front or read off the stream; tools run in this chat's desktop scope. */
+  chatId?: string
+}
+
+/** Relays by stream id. An entry is removed when its relay ends. */
+const relays = new Map<string, Relay>()
+
+/** The call a tool result frame settles, or undefined for any other event. */
+function settledToolCallId(event: PersistedStreamEventEnvelope): string | undefined {
+  if (event.type !== MothershipStreamV1EventType.tool || 'previewPhase' in event.payload) {
+    return undefined
+  }
+  return event.payload.phase === MothershipStreamV1ToolPhase.result
+    ? event.payload.toolCallId
+    : undefined
+}
+
+/**
+ * Starts the client tools a detached turn hands this client. Workflow runs are
+ * left alone: running one drives the workflow editor, and the server runs a
+ * workflow call itself when no client picks it up.
+ */
+function startDetachedClientTool(
+  turn: DetachedChatTurn,
+  chatId: string,
+  start: ClientToolStart
+): void {
+  const { toolCallId, toolName, args, eventTs } = start
+  const scopeId = desktopChatScopeId(turn.scopeKey, chatId)
+  switch (start.kind) {
+    case 'workflow':
+      return
+    case 'localFilesystem':
+      launchLocalFilesystemTool(toolCallId, toolName, args, {
+        workspaceId: turn.workspaceId,
+        chatId,
+      })
+      return
+    case 'browser':
+      executeBrowserToolOnClient(toolCallId, start.toolName, args, scopeId, eventTs)
+      return
+    case 'terminal':
+      executeTerminalToolOnClient(toolCallId, args, scopeId, eventTs)
+      return
+  }
+}
+
+/**
+ * Relays until the run ends or the relay is aborted. Like the chat view's own
+ * reconnect, every connection first reads the events past the cursor as one
+ * batch, so calls that already have a result are settled before any call frame
+ * replays, then tails live. That makes any cursor a safe starting point.
+ */
+async function relayClientTools(turn: DetachedChatTurn, relay: Relay): Promise<void> {
+  const { streamId } = turn
+  const { signal } = relay.controller
+  const headers = turn.traceparent ? { traceparent: turn.traceparent } : undefined
+  /** Calls this relay started or saw settle; ids alone decide what is pending. */
+  const handledToolCallIds = new Set<string>()
+  let cursor = turn.afterCursor
+  let failedAttempts = 0
+
+  const applyEvent = (event: PersistedStreamEventEnvelope): void => {
+    const eventCursor = getStreamEventCursor(event)
+    if (isAlreadyProcessedStreamCursor(eventCursor, cursor)) return
+    cursor = eventCursor
+    relay.chatId ??= resolveChatIdFromStreamEvent(event)
+    if (event.type !== MothershipStreamV1EventType.tool) return
+    const settledId = settledToolCallId(event)
+    if (settledId) {
+      handledToolCallIds.add(settledId)
+      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', {
+        streamId,
+        toolCallId: start.toolCallId,
+      })
+      return
+    }
+    startDetachedClientTool(turn, relay.chatId, start)
+  }
+
+  /** 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: AbortSignal.any([signal, AbortSignal.timeout(STREAM_BATCH_FETCH_TIMEOUT_MS)]),
+      headers,
+    })
+    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)
+    for (const { event } of batch.events) {
+      const settledId = settledToolCallId(event)
+      if (settledId) handledToolCallIds.add(settledId)
+    }
+    for (const { event } of batch.events) applyEvent(event)
+    return isTerminalStreamStatus(batch.status)
+  }
+
+  /** 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 (isRelayEndStatus(response.status)) return true
+    if (!response.ok || !response.body) {
+      throw new Error(`Stream tail responded with status ${response.status}`)
+    }
+    let complete = false
+    await readSSELines(response.body, {
+      signal,
+      idleTimeoutMs: STREAM_IDLE_TIMEOUT_MS,
+      onData: (raw) => {
+        const parsed = parsePersistedStreamEventEnvelopeJson(raw)
+        if (!parsed.ok) throw createStreamSchemaValidationError(parsed, 'Detached SSE event.')
+        applyEvent(parsed.event)
+        if (parsed.event.type === MothershipStreamV1EventType.complete) {
+          complete = true
+          return true
+        }
+      },
+    })
+    return complete
+  }
+
+  while (!signal.aborted) {
+    const cursorBeforeAttempt = cursor
+    try {
+      if ((await readBatch()) || (await readTail())) return
+    } catch (error) {
+      if (signal.aborted) return
+      if (isStreamSchemaValidationError(error)) {
+        logger.error('Stopped relaying detached client tools on an invalid stream event', {
+          streamId,
+          error: error.message,
+        })
+        return
+      }
+      logger.warn('Detached stream read failed', { streamId, error: getErrorMessage(error) })
+    }
+    if (cursor !== cursorBeforeAttempt) {
+      failedAttempts = 0
+      continue
+    }
+    failedAttempts++
+    await interruptibleSleep(backoffWithJitter(failedAttempts, null), signal)
+  }
+}
+
+/** Relays a left turn's client tools until its run ends or its chat's view reads it again. */
+export function detachClientTools(turn: DetachedChatTurn): void {
+  relays.get(turn.streamId)?.controller.abort('superseded_detached_relay')
+  const relay: Relay = { controller: new AbortController(), chatId: turn.chatId }
+  relays.set(turn.streamId, relay)
+  void relayClientTools(turn, relay).finally(() => {
+    if (relays.get(turn.streamId) === relay) relays.delete(turn.streamId)
+  })
+}
+
+/**
+ * Stops relaying a stream the chat view reads again. Tools the relay already
+ * started keep running and report their outcome.
+ */
+export function reattachClientTools(streamId: string): void {
+  relays.get(streamId)?.controller.abort('chat_reattached')
+  relays.delete(streamId)
+}