From f9df0e82497eba858aac3cdb19506796a154f229 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Fri, 25 Sep 2026 16:19:36 -0700 Subject: [PATCH 1/3] fix(chat): keep desktop tools running in chats left mid-turn --- .../home/hooks/stream-protocol.ts | 22 +++ .../home/hooks/stream/client-tool-start.ts | 69 +++++++ .../hooks/stream/detached-client-tools.ts | 187 ++++++++++++++++++ .../home/hooks/stream/handle-tool-event.ts | 104 ++++------ .../[workspaceId]/home/hooks/use-chat.ts | 142 +++++++------ .../client/launch-local-filesystem-tool.ts | 66 +++++++ .../tools/client/local-filesystem.ts | 2 +- 7 files changed, 447 insertions(+), 145 deletions(-) create mode 100644 apps/sim/app/workspace/[workspaceId]/home/hooks/stream/client-tool-start.ts create mode 100644 apps/sim/app/workspace/[workspaceId]/home/hooks/stream/detached-client-tools.ts create mode 100644 apps/sim/lib/mothership/tools/client/launch-local-filesystem-tool.ts diff --git a/apps/sim/app/workspace/[workspaceId]/home/hooks/stream-protocol.ts b/apps/sim/app/workspace/[workspaceId]/home/hooks/stream-protocol.ts index c27b6a28930..5302b348b6b 100644 --- a/apps/sim/app/workspace/[workspaceId]/home/hooks/stream-protocol.ts +++ b/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,9 @@ 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 type StreamBatchResponse = { success: boolean events: StreamBatchEvent[] @@ -148,6 +152,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 diff --git a/apps/sim/app/workspace/[workspaceId]/home/hooks/stream/client-tool-start.ts b/apps/sim/app/workspace/[workspaceId]/home/hooks/stream/client-tool-start.ts new file mode 100644 index 00000000000..b81fd5406c7 --- /dev/null +++ b/apps/sim/app/workspace/[workspaceId]/home/hooks/stream/client-tool-start.ts @@ -0,0 +1,69 @@ +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 + +interface ClientToolCall { + toolCallId: string + args: Record + /** 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 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 + */ +export function resolveClientToolStart( + event: ToolEvent, + isPending: (toolCallId: string) => boolean +): 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 | 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 } + } + return start && isPending(toolCallId) ? start : null +} diff --git a/apps/sim/app/workspace/[workspaceId]/home/hooks/stream/detached-client-tools.ts b/apps/sim/app/workspace/[workspaceId]/home/hooks/stream/detached-client-tools.ts new file mode 100644 index 00000000000..3284274dde8 --- /dev/null +++ b/apps/sim/app/workspace/[workspaceId]/home/hooks/stream/detached-client-tools.ts @@ -0,0 +1,187 @@ +/** + * 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 tails + * the left chat's stream headlessly and starts those tools until the run + * completes or the chat is reopened. Relays live at module scope because + * switching chats remounts the chat surface that detached them. Each holds one + * resume connection, so a relay exists only while its left 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 { + MothershipStreamV1EventType, + MothershipStreamV1ToolPhase, +} from '@/lib/mothership/generated/mothership-stream-v1' +import { 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, + STREAM_IDLE_TIMEOUT_MS, +} from '@/app/workspace/[workspaceId]/home/hooks/stream-protocol' + +const logger = createLogger('DetachedClientTools') + +/** Consecutive tail attempts that make no progress before a relay gives up. */ +const MAX_STALLED_TAIL_ATTEMPTS = 5 + +/** A live turn the user left, and where its client tools run. */ +export interface DetachedChatTurn { + chatId: string + streamId: string + /** The last cursor the chat view dispatched; the relay resumes after it. */ + afterCursor: string + traceparent?: string + workspaceId?: string + /** The chat's desktop scope, which owns its browser and terminal tabs. */ + scopeId: string +} + +/** Relays by chat id. An entry is removed when its relay ends. */ +const relays = new Map() + +/** + * 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, start: ClientToolStart): void { + const { toolCallId, toolName, args, eventTs } = start + switch (start.kind) { + case 'workflow': + return + case 'localFilesystem': + launchLocalFilesystemTool(toolCallId, toolName, args, { + workspaceId: turn.workspaceId, + chatId: turn.chatId, + }) + return + case 'browser': + executeBrowserToolOnClient(toolCallId, start.toolName, args, turn.scopeId, eventTs) + return + case 'terminal': + executeTerminalToolOnClient(toolCallId, args, turn.scopeId, eventTs) + return + } +} + +async function relayClientTools(turn: DetachedChatTurn, signal: AbortSignal): Promise { + const { streamId, traceparent } = turn + /** Calls this relay started or saw settle; ids alone decide what is pending. */ + const handledToolCallIds = new Set() + let cursor = turn.afterCursor + let stalledAttempts = 0 + + /** Reads one tail connection; resolves true once the run is over. */ + const readTail = async (): Promise => { + // boundary-raw-fetch: live SSE tail endpoint streams events consumed via readSSELines + const response = await fetch(buildStreamResumeUrl(streamId, cursor), { + signal, + ...(traceparent ? { headers: { traceparent } } : {}), + }) + if (response.status === 404) 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.') + const event = parsed.event + const eventCursor = getStreamEventCursor(event) + if (isAlreadyProcessedStreamCursor(eventCursor, cursor)) return + cursor = eventCursor + + if (event.type === MothershipStreamV1EventType.tool) { + const start = resolveClientToolStart(event, (id) => !handledToolCallIds.has(id)) + if (start) { + handledToolCallIds.add(start.toolCallId) + startDetachedClientTool(turn, start) + } else if ( + !('previewPhase' in event.payload) && + event.payload.phase === MothershipStreamV1ToolPhase.result + ) { + handledToolCallIds.add(event.payload.toolCallId) + } + } + if (event.type === MothershipStreamV1EventType.complete) { + complete = true + return true + } + }, + }) + return complete + } + + while (!signal.aborted) { + const cursorBeforeAttempt = cursor + try { + if (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 tail failed', { streamId, error: getErrorMessage(error) }) + } + + if (cursor !== cursorBeforeAttempt) { + stalledAttempts = 0 + continue + } + stalledAttempts++ + if (stalledAttempts >= MAX_STALLED_TAIL_ATTEMPTS) { + logger.warn('Stopped relaying detached client tools after repeated stalls', { + streamId, + cursor, + }) + return + } + await interruptibleSleep(backoffWithJitter(stalledAttempts, null), signal) + } +} + +/** + * Relays a left turn's client tools until its run completes or the chat is + * reopened. Replaces any relay the chat already has. + */ +export function detachClientTools(turn: DetachedChatTurn): void { + relays.get(turn.chatId)?.abort('superseded_detached_relay') + const controller = new AbortController() + relays.set(turn.chatId, controller) + void relayClientTools(turn, controller.signal).finally(() => { + if (relays.get(turn.chatId) === controller) relays.delete(turn.chatId) + }) +} + +/** + * Stops relaying a chat the user reopened; its chat view takes the turn back. + * Tools the relay already started keep running and report their outcome. + */ +export function reattachClientTools(chatId: string): void { + relays.get(chatId)?.abort('chat_reattached') + relays.delete(chatId) +} diff --git a/apps/sim/app/workspace/[workspaceId]/home/hooks/stream/handle-tool-event.ts b/apps/sim/app/workspace/[workspaceId]/home/hooks/stream/handle-tool-event.ts index 0fb9877fa43..08d42463086 100644 --- a/apps/sim/app/workspace/[workspaceId]/home/hooks/stream/handle-tool-event.ts +++ b/apps/sim/app/workspace/[workspaceId]/home/hooks/stream/handle-tool-event.ts @@ -1,23 +1,23 @@ -import { isCurrentBrowserToolName } from '@sim/browser-protocol' -import { isTerminalToolName } from '@sim/terminal-protocol' -import { - MothershipStreamV1ToolPhase, - MothershipStreamV1ToolStatus, -} from '@/lib/mothership/generated/mothership-stream-v1' +import { MothershipStreamV1ToolPhase } from '@/lib/mothership/generated/mothership-stream-v1' import { ApplyFileEdit, ConnectSlackBot, PrepareFileEdit, } from '@/lib/mothership/generated/tool-catalog-v1' -import type { PersistedStreamEventEnvelope } from '@/lib/mothership/request/session/contract' import { extractResourcesFromToolResult, isResourceToolName, } from '@/lib/mothership/resources/extraction' -import { isNativeFileTool, isUserLocalVfsToolCall } from '@/lib/mothership/tools/local-filesystem' -import { isWorkflowToolName } from '@/lib/mothership/tools/workflow-tools' import { invalidateResourceQueries } from '@/app/workspace/[workspaceId]/home/components/mothership-view/components/resource-registry' -import type { StreamLoopContext } from '@/app/workspace/[workspaceId]/home/hooks/stream/stream-context' +import { + type ClientToolStart, + resolveClientToolStart, + type ToolEvent, +} from '@/app/workspace/[workspaceId]/home/hooks/stream/client-tool-start' +import type { + StreamLoopContext, + StreamLoopDeps, +} from '@/app/workspace/[workspaceId]/home/hooks/stream/stream-context' import { DEPLOY_TOOL_NAMES, FILE_SUBAGENT_ID, @@ -37,8 +37,6 @@ import { folderKeys } from '@/hooks/queries/utils/folder-keys' import { invalidateWorkflowLists } from '@/hooks/queries/utils/invalidate-workflow-lists' import { invalidateSelectorQueries } from '@/hooks/queries/utils/selector-keys' -type ToolEvent = Extract - /** The display agent id for a tool's owning span (undefined on the main lane). */ function agentIdForSpan(ctx: StreamLoopContext, spanId: string): string | undefined { if (spanId === MAIN_SPAN) return undefined @@ -148,10 +146,29 @@ function runToolResultSideEffects(ctx: StreamLoopContext, node: ToolNode, replay } } +/** Hands a client-executed tool call to the executor for its kind. */ +function startClientTool(deps: StreamLoopDeps, start: ClientToolStart): void { + const { toolCallId, toolName, args, eventTs } = start + switch (start.kind) { + case 'workflow': + deps.startClientWorkflowTool(toolCallId, toolName, args) + return + case 'localFilesystem': + deps.startClientLocalFilesystemTool(toolCallId, toolName, args) + return + case 'browser': + deps.startClientBrowserTool(toolCallId, toolName, args, eventTs) + return + case 'terminal': + deps.startClientTerminalTool(toolCallId, toolName, args, eventTs) + return + } +} + /** * Side effects for tool events. State (the tool node, its status, args, and the * apply_file_edit row merge) is owned by `reduceEvent`; this handler routes preview - * phases, fires client workflow tools, and runs result side effects, then + * phases, starts client-executed tools, and runs result side effects, then * flushes the model-derived snapshot. */ export function handleToolEvent(ctx: StreamLoopContext, parsed: ToolEvent): void { @@ -185,60 +202,11 @@ 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 name = payload.toolName - const isPartial = - payload.partial === true || payload.status === MothershipStreamV1ToolStatus.generating - if (isWorkflowToolName(name) && !isPartial) { - const shouldStartWorkflowTool = - !deps.options.suppressedWorkflowToolStartIds?.has(rawId) && - node?.kind === 'tool' && - node.status === 'running' && - !node.result - if (shouldStartWorkflowTool) { - const args = payload.arguments as Record | undefined - deps.startClientWorkflowTool(rawId, name, args ?? {}) - } - } - const localFilesystemArgs = payload.arguments as Record | undefined - if ((isNativeFileTool(name) || isUserLocalVfsToolCall(name, localFilesystemArgs)) && !isPartial) { - const shouldStartLocalFilesystemTool = - !deps.options.suppressedWorkflowToolStartIds?.has(rawId) && - node?.kind === 'tool' && - node.status === 'running' && - !node.result - if (shouldStartLocalFilesystemTool) { - deps.startClientLocalFilesystemTool(rawId, name, localFilesystemArgs ?? {}) - } - } - if (isCurrentBrowserToolName(name) && !isPartial) { - const shouldStartBrowserTool = - !deps.options.suppressedWorkflowToolStartIds?.has(rawId) && - node?.kind === 'tool' && - node.status === 'running' && - !node.result - if (shouldStartBrowserTool) { - deps.startClientBrowserTool( - rawId, - name, - (payload.arguments as Record | undefined) ?? {}, - parsed.ts - ) - } - } - if (isTerminalToolName(name) && !isPartial) { - const shouldStartTerminalTool = - !deps.options.suppressedWorkflowToolStartIds?.has(rawId) && - node?.kind === 'tool' && - node.status === 'running' && - !node.result - if (shouldStartTerminalTool) { - deps.startClientTerminalTool( - rawId, - name, - (payload.arguments as Record | undefined) ?? {}, - parsed.ts - ) - } - } + 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) ops.flush() } diff --git a/apps/sim/app/workspace/[workspaceId]/home/hooks/use-chat.ts b/apps/sim/app/workspace/[workspaceId]/home/hooks/use-chat.ts index 15de9ecf2f0..c79fa65bfd3 100644 --- a/apps/sim/app/workspace/[workspaceId]/home/hooks/use-chat.ts +++ b/apps/sim/app/workspace/[workspaceId]/home/hooks/use-chat.ts @@ -34,7 +34,7 @@ import { getDeploymentShape } from '@/lib/core/config/deployment-shape' import { MothershipHandoffStorage } from '@/lib/core/utils/browser-storage' import { withinDeadline } from '@/lib/core/utils/deadline' import { readSSELines } from '@/lib/core/utils/sse' -import { getDesktopBridge, getDesktopChatCapabilities } from '@/lib/desktop' +import { getDesktopBridge, getDesktopChatCapabilities, isDesktopApp } from '@/lib/desktop' import { activateDesktopChatScopes, desktopChatScopeId, @@ -75,6 +75,7 @@ import { sanitizeChatResources, } from '@/lib/mothership/resources/types' import { executeBrowserToolOnClient } from '@/lib/mothership/tools/client/browser-tool-execution' +import { launchLocalFilesystemTool } from '@/lib/mothership/tools/client/launch-local-filesystem-tool' import { bindRunToolToExecution, executeRunToolOnClient, @@ -82,7 +83,6 @@ import { } from '@/lib/mothership/tools/client/run-tool-execution' import { executeTerminalToolOnClient } from '@/lib/mothership/tools/client/terminal-tool-execution' import { setCurrentChatTraceparent } from '@/lib/mothership/tools/client/trace-context' -import { isNativeFileTool, isUserLocalVfsToolCall } from '@/lib/mothership/tools/local-filesystem' import { isWorkflowToolName } from '@/lib/mothership/tools/workflow-tools' import { initTerminalTransport } from '@/lib/terminal/transport' import { getQueryClient } from '@/app/_shell/providers/get-query-client' @@ -104,6 +104,10 @@ import { dispatchStreamEvent, finalizeResidualToolCalls, } from '@/app/workspace/[workspaceId]/home/hooks/stream' +import { + detachClientTools, + reattachClientTools, +} from '@/app/workspace/[workspaceId]/home/hooks/stream/detached-client-tools' import { useNativeActiveTabIds } from '@/app/workspace/[workspaceId]/home/hooks/use-desktop-tab-resources' import { resolveEffectiveResourceId } from '@/app/workspace/[workspaceId]/home/resource-view-policy' import { useFeatureFlag } from '@/app/workspace/[workspaceId]/providers/feature-flags-provider' @@ -166,12 +170,16 @@ import { } from './send-handoff' import { buildReplayStream, + buildStreamResumeUrl, createStreamSchemaValidationError, + getStreamEventCursor, isAlreadyProcessedStreamCursor, isStreamGoneError, isStreamSchemaValidationError, + isZeroStreamCursor, parseStreamBatchResponse, resolveChatIdFromStreamBatch, + STREAM_IDLE_TIMEOUT_MS, type StreamBatchResponse, StreamGoneError, } from './stream-protocol' @@ -287,8 +295,6 @@ 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 -/** Both live transports heartbeat every 15s; three missed heartbeats trigger cursor recovery. */ -const STREAM_IDLE_TIMEOUT_MS = 45_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 @@ -905,6 +911,7 @@ export function useChat( (reason: 'pageshow' | 'visible' | 'online' | 'exhausted_recheck') => Promise >(async () => {}) const reconnectExhaustedRecheckTimerRef = useRef | null>(null) + const detachLiveTurnClientToolsRef = useRef<() => boolean>(() => false) const abortControllerRef = useRef(null) const detachedChatResolutionControllersRef = useRef>(new Set()) @@ -960,7 +967,6 @@ export function useChat( const streamingContentRef = useRef('') const streamingBlocksRef = useRef([]) const handledClientWorkflowToolIdsRef = useRef>(new Set()) - const handledClientLocalFilesystemToolIdsRef = useRef>(new Set()) const recoveringClientWorkflowToolIdsRef = useRef>(new Set()) const isHomePage = pathname.endsWith('/home') @@ -1043,6 +1049,7 @@ export function useChat( const resetHomeChatState = useCallback(() => { const abandonedDesktopScopeId = desktopScopeIdRef.current + detachLiveTurnClientToolsRef.current() cancelActiveStreamRecovery() streamGenRef.current++ cancelActiveStreamReader() @@ -1530,61 +1537,13 @@ export function useChat( const startClientLocalFilesystemTool = useCallback( (toolCallId: string, toolName: string, toolArgs: Record) => { - if ( - !isNativeFileTool(toolName) && - (!workspaceId || !isUserLocalVfsToolCall(toolName, toolArgs)) - ) { - return - } - if (handledClientLocalFilesystemToolIdsRef.current.has(toolCallId)) { - return - } - handledClientLocalFilesystemToolIdsRef.current.add(toolCallId) - const options = { + launchLocalFilesystemTool(toolCallId, toolName, toolArgs, { workspaceId, chatId: chatIdRef.current ?? selectedChatIdRef.current, signal: abortControllerRef.current?.signal, - } - /** - * Dynamic on purpose: the local-filesystem executor only runs for desktop-local - * VFS tool calls, and a static import kept it in the shared chat chunk on every - * surface that mounts the composer. The guard, the dedupe add, and the option - * capture above stay synchronous, so re-entrancy behaviour is unchanged. If the - * chunk fails to load (deploy skew), the server-side tool call must still settle: - * report an error completion rather than leaving it hanging with the dedupe ref - * already marked handled. - */ - import('@/lib/mothership/tools/client/local-filesystem').then( - (m) => m.executeLocalFilesystemTool(toolCallId, toolName, toolArgs, options), - async (error) => { - logger.error('Failed to load local filesystem tool executor', { error }) - /** - * The recovery itself can reject (the helper chunks or the completion POST can - * fail for the same reason the executor chunk did). Contain it: an unhandled - * rejection here would settle nothing and surface as a console error, exactly - * like the executor's own report-failure path, which also degrades to a log. - */ - try { - const [{ reportClientToolCompletion }, { ASYNC_TOOL_CONFIRMATION_STATUS }] = - await Promise.all([ - import('@/lib/mothership/tools/client/completion'), - import('@/lib/mothership/async-runs/lifecycle'), - ]) - await reportClientToolCompletion( - toolCallId, - ASYNC_TOOL_CONFIRMATION_STATUS.error, - 'Local filesystem tool failed to load' - ) - } catch (reportError) { - logger.error('Failed to report local filesystem tool load failure', { - toolCallId, - error: reportError, - }) - } - } - ) + }) }, - [workspaceId, organizationId, scopeKey] + [workspaceId] ) const getResourceActivityTracker = useCallback( @@ -1649,6 +1608,40 @@ export function useChat( [workspaceId, organizationId, scopeKey] ) + /** + * 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. When it does, + * the caller releases the turn's controller without aborting it: tools + * already running report their outcome instead of being cancelled. A view + * that has applied no event yet cannot tell settled calls from pending ones + * on replay, so it keeps no relay. + * + * @returns whether a relay took the turn + */ + detachLiveTurnClientToolsRef.current = (): boolean => { + const chatId = chatIdRef.current + const streamId = streamIdRef.current + if ( + !isDesktopApp() || + requestModeRef.current === 'assistant' || + !sendingRef.current || + !chatId || + !streamId || + isZeroStreamCursor(lastCursorRef.current) + ) { + return false + } + detachClientTools({ + chatId, + streamId, + afterCursor: lastCursorRef.current, + traceparent: streamTraceparentRef.current, + workspaceId, + scopeId: desktopChatScopeId(scopeKey, chatId), + }) + return true + } + const recoverPendingClientWorkflowTools = useCallback( async (nextMessages: ChatMessage[]) => { const pending: ToolCallInfo[] = [] @@ -1693,6 +1686,7 @@ export function useChat( ) useEffect(() => { + if (initialChatId) reattachClientTools(initialChatId) const previousDesktopScopeId = desktopScopeIdRef.current const canDiscardPreviousPendingScope = !sendingRef.current const streamOwnerId = chatIdRef.current @@ -1767,6 +1761,7 @@ export function useChat( } // Detach the current UI from the old stream without cancelling it on the server. // Reopening that chat later will reconnect through the existing chatHistory flow. + detachLiveTurnClientToolsRef.current() cancelActiveStreamRecovery() streamGenRef.current++ cancelActiveStreamReader() @@ -2247,7 +2242,7 @@ export function useChat( if (parsed.stream?.streamId) { streamIdRef.current = parsed.stream.streamId } - const eventCursor = parsed.stream?.cursor ?? String(parsed.seq) + const eventCursor = getStreamEventCursor(parsed) if (isAlreadyProcessedStreamCursor(eventCursor, lastCursorRef.current)) { return } @@ -2366,15 +2361,12 @@ export function useChat( createTimeoutSignal(STREAM_BATCH_FETCH_TIMEOUT_MS) ) // boundary-raw-fetch: stream-resume batch endpoint requires dynamic per-request traceparent header propagation that the contract layer does not model, and the response is consumed alongside live SSE tail fetches - const response = await fetch( - `/api/mothership/chat/stream?streamId=${encodeURIComponent(streamId)}&after=${encodeURIComponent(afterCursor)}&batch=true`, - { - signal: fetchSignal, - ...(streamTraceparentRef.current - ? { headers: { traceparent: streamTraceparentRef.current } } - : {}), - } - ) + const response = await fetch(buildStreamResumeUrl(streamId, afterCursor, { batch: true }), { + signal: fetchSignal, + ...(streamTraceparentRef.current + ? { headers: { traceparent: streamTraceparentRef.current } } + : {}), + }) if (response.status === 404) { throw new StreamGoneError(streamId) } @@ -2558,15 +2550,12 @@ export function useChat( logger.info('Opening live stream tail', { streamId, afterCursor: latestCursor }) // boundary-raw-fetch: live SSE tail endpoint streams events consumed via response.body.getReader() and processSSEStream - const sseRes = await fetch( - `/api/mothership/chat/stream?streamId=${encodeURIComponent(streamId)}&after=${encodeURIComponent(latestCursor)}`, - { - signal: activeAbort.signal, - ...(streamTraceparentRef.current - ? { headers: { traceparent: streamTraceparentRef.current } } - : {}), - } - ) + const sseRes = await fetch(buildStreamResumeUrl(streamId, latestCursor), { + signal: activeAbort.signal, + ...(streamTraceparentRef.current + ? { headers: { traceparent: streamTraceparentRef.current } } + : {}), + }) if (sseRes.status === 404) { throw new StreamGoneError(streamId) } @@ -4892,11 +4881,12 @@ export function useChat( useEffect(() => { return () => { + const detachedToRelay = detachLiveTurnClientToolsRef.current() cancelActiveStreamRecovery() clearQueueDispatchState() streamGenRef.current++ cancelActiveStreamReader() - abortControllerRef.current?.abort('unmount:client_cleanup') + if (!detachedToRelay) abortControllerRef.current?.abort('unmount:client_cleanup') abortControllerRef.current = null for (const controller of detachedChatResolutionControllersRef.current) { controller.abort('unmount:detached_chat_resolution') diff --git a/apps/sim/lib/mothership/tools/client/launch-local-filesystem-tool.ts b/apps/sim/lib/mothership/tools/client/launch-local-filesystem-tool.ts new file mode 100644 index 00000000000..b9059f674a9 --- /dev/null +++ b/apps/sim/lib/mothership/tools/client/launch-local-filesystem-tool.ts @@ -0,0 +1,66 @@ +import { createLogger } from '@sim/logger' +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. + */ +const launchedToolCallIds = new Set() + +/** + * 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 + * kept it in the shared chat chunk on every surface that mounts the composer. + * If the chunk fails to load (deploy skew), the server-side waiter must still + * settle, so the failure is reported as an error completion instead of leaving + * the call hanging. + */ +export function launchLocalFilesystemTool( + toolCallId: string, + toolName: string, + args: Record, + context: LocalFilesystemExecutionContext +): void { + if ( + !isNativeFileTool(toolName) && + (!context.workspaceId || !isUserLocalVfsToolCall(toolName, args)) + ) { + return + } + if (launchedToolCallIds.has(toolCallId)) return + launchedToolCallIds.add(toolCallId) + + import('@/lib/mothership/tools/client/local-filesystem').then( + (m) => m.executeLocalFilesystemTool(toolCallId, toolName, args, context), + async (error) => { + logger.error('Failed to load local filesystem tool executor', { error }) + /** + * The recovery itself can reject (the helper chunks or the completion POST can + * fail for the same reason the executor chunk did). Contain it: an unhandled + * rejection here would settle nothing and surface as a console error, exactly + * like the executor's own report-failure path, which also degrades to a log. + */ + try { + const [{ reportClientToolCompletion }, { ASYNC_TOOL_CONFIRMATION_STATUS }] = + await Promise.all([ + import('@/lib/mothership/tools/client/completion'), + import('@/lib/mothership/async-runs/lifecycle'), + ]) + await reportClientToolCompletion( + toolCallId, + ASYNC_TOOL_CONFIRMATION_STATUS.error, + 'Local filesystem tool failed to load' + ) + } catch (reportError) { + logger.error('Failed to report local filesystem tool load failure', { + toolCallId, + error: reportError, + }) + } + } + ) +} diff --git a/apps/sim/lib/mothership/tools/client/local-filesystem.ts b/apps/sim/lib/mothership/tools/client/local-filesystem.ts index 65760975187..db33953eee3 100644 --- a/apps/sim/lib/mothership/tools/client/local-filesystem.ts +++ b/apps/sim/lib/mothership/tools/client/local-filesystem.ts @@ -38,7 +38,7 @@ const VFS_GLOB_OPTIONS: micromatch.Options = { noext: true, } -interface LocalFilesystemExecutionContext { +export interface LocalFilesystemExecutionContext { workspaceId?: string chatId?: string signal?: AbortSignal From 4af9b7a265d3d1629a1aeb78e66ecd904911038f Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Fri, 25 Sep 2026 16:36:39 -0700 Subject: [PATCH 2/3] fix(chat): relay left turns from any cursor and until the run ends --- .../home/hooks/stream-protocol.ts | 25 ++- .../hooks/stream/detached-client-tools.ts | 193 +++++++++++------- .../[workspaceId]/home/hooks/use-chat.ts | 27 ++- .../client/launch-local-filesystem-tool.ts | 8 +- 4 files changed, 156 insertions(+), 97 deletions(-) diff --git a/apps/sim/app/workspace/[workspaceId]/home/hooks/stream-protocol.ts b/apps/sim/app/workspace/[workspaceId]/home/hooks/stream-protocol.ts index 5302b348b6b..114cac039cc 100644 --- a/apps/sim/app/workspace/[workspaceId]/home/hooks/stream-protocol.ts +++ b/apps/sim/app/workspace/[workspaceId]/home/hooks/stream-protocol.ts @@ -115,18 +115,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 diff --git a/apps/sim/app/workspace/[workspaceId]/home/hooks/stream/detached-client-tools.ts b/apps/sim/app/workspace/[workspaceId]/home/hooks/stream/detached-client-tools.ts index 3284274dde8..16a959cc469 100644 --- a/apps/sim/app/workspace/[workspaceId]/home/hooks/stream/detached-client-tools.ts +++ b/apps/sim/app/workspace/[workspaceId]/home/hooks/stream/detached-client-tools.ts @@ -3,22 +3,27 @@ * 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 tails - * the left chat's stream headlessly and starts those tools until the run - * completes or the chat is reopened. Relays live at module scope because - * switching chats remounts the chat surface that detached them. Each holds one - * resume connection, so a relay exists only while its left run is live. + * 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 { parsePersistedStreamEventEnvelopeJson } from '@/lib/mothership/request/session/contract' +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' @@ -32,73 +37,141 @@ import { getStreamEventCursor, isAlreadyProcessedStreamCursor, isStreamSchemaValidationError, + parseStreamBatchResponse, + resolveChatIdFromStreamBatch, + resolveChatIdFromStreamEvent, STREAM_IDLE_TIMEOUT_MS, } from '@/app/workspace/[workspaceId]/home/hooks/stream-protocol' const logger = createLogger('DetachedClientTools') -/** Consecutive tail attempts that make no progress before a relay gives up. */ -const MAX_STALLED_TAIL_ATTEMPTS = 5 - -/** A live turn the user left, and where its client tools run. */ +/** A live turn the user left. */ export interface DetachedChatTurn { - chatId: string 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 chat's desktop scope, which owns its browser and terminal tabs. */ - scopeId: string + /** The owner key the chat's desktop scope derives from. */ + scopeKey: string } -/** Relays by chat id. An entry is removed when its relay ends. */ -const relays = new Map() +interface Relay { + controller: AbortController + chatId?: string +} + +/** Relays by stream id. An entry is removed when its relay ends. */ +const relays = new Map() + +/** 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, start: ClientToolStart): void { +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: turn.chatId, + chatId, }) return case 'browser': - executeBrowserToolOnClient(toolCallId, start.toolName, args, turn.scopeId, eventTs) + executeBrowserToolOnClient(toolCallId, start.toolName, args, scopeId, eventTs) return case 'terminal': - executeTerminalToolOnClient(toolCallId, args, turn.scopeId, eventTs) + executeTerminalToolOnClient(toolCallId, args, scopeId, eventTs) return } } -async function relayClientTools(turn: DetachedChatTurn, signal: AbortSignal): Promise { - const { streamId, traceparent } = turn +/** + * 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 { + 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() let cursor = turn.afterCursor - let stalledAttempts = 0 + let failedAttempts = 0 - /** Reads one tail connection; resolves true once the run is over. */ - const readTail = async (): Promise => { - // boundary-raw-fetch: live SSE tail endpoint streams events consumed via readSSELines - const response = await fetch(buildStreamResumeUrl(streamId, cursor), { + 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, (id) => !handledToolCallIds.has(id)) + if (!start) 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 the run is over. */ + const readBatch = async (): Promise => { + // 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, - ...(traceparent ? { headers: { traceparent } } : {}), + headers, }) if (response.status === 404) 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 the run is over. */ + const readTail = async (): Promise => { + // 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 (!response.ok || !response.body) { throw new Error(`Stream tail responded with status ${response.status}`) } - let complete = false await readSSELines(response.body, { signal, @@ -106,24 +179,8 @@ async function relayClientTools(turn: DetachedChatTurn, signal: AbortSignal): Pr onData: (raw) => { const parsed = parsePersistedStreamEventEnvelopeJson(raw) if (!parsed.ok) throw createStreamSchemaValidationError(parsed, 'Detached SSE event.') - const event = parsed.event - const eventCursor = getStreamEventCursor(event) - if (isAlreadyProcessedStreamCursor(eventCursor, cursor)) return - cursor = eventCursor - - if (event.type === MothershipStreamV1EventType.tool) { - const start = resolveClientToolStart(event, (id) => !handledToolCallIds.has(id)) - if (start) { - handledToolCallIds.add(start.toolCallId) - startDetachedClientTool(turn, start) - } else if ( - !('previewPhase' in event.payload) && - event.payload.phase === MothershipStreamV1ToolPhase.result - ) { - handledToolCallIds.add(event.payload.toolCallId) - } - } - if (event.type === MothershipStreamV1EventType.complete) { + applyEvent(parsed.event) + if (parsed.event.type === MothershipStreamV1EventType.complete) { complete = true return true } @@ -135,7 +192,7 @@ async function relayClientTools(turn: DetachedChatTurn, signal: AbortSignal): Pr while (!signal.aborted) { const cursorBeforeAttempt = cursor try { - if (await readTail()) return + if ((await readBatch()) || (await readTail())) return } catch (error) { if (signal.aborted) return if (isStreamSchemaValidationError(error)) { @@ -145,43 +202,35 @@ async function relayClientTools(turn: DetachedChatTurn, signal: AbortSignal): Pr }) return } - logger.warn('Detached stream tail failed', { streamId, error: getErrorMessage(error) }) + logger.warn('Detached stream read failed', { streamId, error: getErrorMessage(error) }) } - if (cursor !== cursorBeforeAttempt) { - stalledAttempts = 0 + failedAttempts = 0 continue } - stalledAttempts++ - if (stalledAttempts >= MAX_STALLED_TAIL_ATTEMPTS) { - logger.warn('Stopped relaying detached client tools after repeated stalls', { - streamId, - cursor, - }) - return - } - await interruptibleSleep(backoffWithJitter(stalledAttempts, null), signal) + failedAttempts++ + await interruptibleSleep(backoffWithJitter(failedAttempts, null), signal) } } -/** - * Relays a left turn's client tools until its run completes or the chat is - * reopened. Replaces any relay the chat already has. - */ +/** 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.chatId)?.abort('superseded_detached_relay') - const controller = new AbortController() - relays.set(turn.chatId, controller) - void relayClientTools(turn, controller.signal).finally(() => { - if (relays.get(turn.chatId) === controller) relays.delete(turn.chatId) + 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 chat the user reopened; its chat view takes the turn back. - * Tools the relay already started keep running and report their outcome. + * Stops relaying a chat whose view reads its stream again. Tools the relay + * already started keep running and report their outcome. */ export function reattachClientTools(chatId: string): void { - relays.get(chatId)?.abort('chat_reattached') - relays.delete(chatId) + for (const [streamId, relay] of relays) { + if (relay.chatId !== chatId) continue + relay.controller.abort('chat_reattached') + relays.delete(streamId) + } } diff --git a/apps/sim/app/workspace/[workspaceId]/home/hooks/use-chat.ts b/apps/sim/app/workspace/[workspaceId]/home/hooks/use-chat.ts index c79fa65bfd3..4693e320a83 100644 --- a/apps/sim/app/workspace/[workspaceId]/home/hooks/use-chat.ts +++ b/apps/sim/app/workspace/[workspaceId]/home/hooks/use-chat.ts @@ -176,7 +176,6 @@ import { isAlreadyProcessedStreamCursor, isStreamGoneError, isStreamSchemaValidationError, - isZeroStreamCursor, parseStreamBatchResponse, resolveChatIdFromStreamBatch, STREAM_IDLE_TIMEOUT_MS, @@ -1610,34 +1609,33 @@ 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. When it does, - * the caller releases the turn's controller without aborting it: tools - * already running report their outcome instead of being cancelled. A view - * that has applied no event yet cannot tell settled calls from pending ones - * on replay, so it keeps no relay. + * 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. * * @returns whether a relay took the turn */ detachLiveTurnClientToolsRef.current = (): boolean => { - const chatId = chatIdRef.current - const streamId = streamIdRef.current + const streamId = + streamIdRef.current ?? + (streamReaderRef.current ? activeTurnRef.current?.userMessageId : undefined) if ( !isDesktopApp() || requestModeRef.current === 'assistant' || !sendingRef.current || - !chatId || - !streamId || - isZeroStreamCursor(lastCursorRef.current) + !streamId ) { return false } detachClientTools({ - chatId, streamId, + chatId: chatIdRef.current, afterCursor: lastCursorRef.current, traceparent: streamTraceparentRef.current, workspaceId, - scopeId: desktopChatScopeId(scopeKey, chatId), + scopeKey, }) return true } @@ -1686,7 +1684,6 @@ export function useChat( ) useEffect(() => { - if (initialChatId) reattachClientTools(initialChatId) const previousDesktopScopeId = desktopScopeIdRef.current const canDiscardPreviousPendingScope = !sendingRef.current const streamOwnerId = chatIdRef.current @@ -2213,6 +2210,8 @@ export function useChat( return { sawStreamError: false, sawComplete: false } } streamReaderRef.current = reader + const readerChatId = options?.targetChatId ?? chatIdRef.current + if (readerChatId) reattachClientTools(readerChatId) try { await readSSELines(reader, { diff --git a/apps/sim/lib/mothership/tools/client/launch-local-filesystem-tool.ts b/apps/sim/lib/mothership/tools/client/launch-local-filesystem-tool.ts index b9059f674a9..ea322181314 100644 --- a/apps/sim/lib/mothership/tools/client/launch-local-filesystem-tool.ts +++ b/apps/sim/lib/mothership/tools/client/launch-local-filesystem-tool.ts @@ -1,4 +1,5 @@ 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' @@ -7,9 +8,10 @@ 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. + * 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 Set() +const launchedToolCallIds = new LRUCache({ max: 500 }) /** * Runs a local filesystem tool call on the desktop client. The executor is @@ -32,7 +34,7 @@ export function launchLocalFilesystemTool( return } if (launchedToolCallIds.has(toolCallId)) return - launchedToolCallIds.add(toolCallId) + launchedToolCallIds.set(toolCallId, true) import('@/lib/mothership/tools/client/local-filesystem').then( (m) => m.executeLocalFilesystemTool(toolCallId, toolName, args, context), From b60c9194fce38f5945988b8d191e880e177ac555 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Fri, 25 Sep 2026 16:51:56 -0700 Subject: [PATCH 3/3] fix(chat): detach only admitted turns and report stale terminal calls --- .../home/hooks/stream-protocol.ts | 1 + .../home/hooks/stream/client-tool-start.ts | 31 ++++++---------- .../hooks/stream/detached-client-tools.ts | 37 +++++++++++-------- .../home/hooks/stream/handle-tool-event.ts | 13 ++++--- .../[workspaceId]/home/hooks/use-chat.ts | 23 ++++++------ .../client/launch-local-filesystem-tool.ts | 11 ------ .../tools/client/local-filesystem.ts | 10 +++++ .../tools/client/terminal-tool-execution.ts | 14 +++++++ 8 files changed, 76 insertions(+), 64 deletions(-) diff --git a/apps/sim/app/workspace/[workspaceId]/home/hooks/stream-protocol.ts b/apps/sim/app/workspace/[workspaceId]/home/hooks/stream-protocol.ts index 114cac039cc..1b3788424c0 100644 --- a/apps/sim/app/workspace/[workspaceId]/home/hooks/stream-protocol.ts +++ b/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 diff --git a/apps/sim/app/workspace/[workspaceId]/home/hooks/stream/client-tool-start.ts b/apps/sim/app/workspace/[workspaceId]/home/hooks/stream/client-tool-start.ts index b81fd5406c7..021a7599f2c 100644 --- a/apps/sim/app/workspace/[workspaceId]/home/hooks/stream/client-tool-start.ts +++ b/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 | 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 } diff --git a/apps/sim/app/workspace/[workspaceId]/home/hooks/stream/detached-client-tools.ts b/apps/sim/app/workspace/[workspaceId]/home/hooks/stream/detached-client-tools.ts index 16a959cc469..e7e285f64bc 100644 --- a/apps/sim/app/workspace/[workspaceId]/home/hooks/stream/detached-client-tools.ts +++ b/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 !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 => { // 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 => { // 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) } diff --git a/apps/sim/app/workspace/[workspaceId]/home/hooks/stream/handle-tool-event.ts b/apps/sim/app/workspace/[workspaceId]/home/hooks/stream/handle-tool-event.ts index 08d42463086..3ca3f1a2368 100644 --- a/apps/sim/app/workspace/[workspaceId]/home/hooks/stream/handle-tool-event.ts +++ b/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() } diff --git a/apps/sim/app/workspace/[workspaceId]/home/hooks/use-chat.ts b/apps/sim/app/workspace/[workspaceId]/home/hooks/use-chat.ts index 4693e320a83..bc91bbc22d4 100644 --- a/apps/sim/app/workspace/[workspaceId]/home/hooks/use-chat.ts +++ b/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 } } diff --git a/apps/sim/lib/mothership/tools/client/launch-local-filesystem-tool.ts b/apps/sim/lib/mothership/tools/client/launch-local-filesystem-tool.ts index ea322181314..f89492bc82d 100644 --- a/apps/sim/lib/mothership/tools/client/launch-local-filesystem-tool.ts +++ b/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({ 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), diff --git a/apps/sim/lib/mothership/tools/client/local-filesystem.ts b/apps/sim/lib/mothership/tools/client/local-filesystem.ts index db33953eee3..5b8d403f7ca 100644 --- a/apps/sim/lib/mothership/tools/client/local-filesystem.ts +++ b/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({ max: 500 }) + export function executeLocalFilesystemTool( toolCallId: string, toolName: string, args: Record, context: LocalFilesystemExecutionContext ): void { + if (executedToolCallIds.has(toolCallId)) return + executedToolCallIds.set(toolCallId, true) if (isNativeFileTool(toolName)) { void executeNativeFileTool(toolCallId, toolName, context.signal) return diff --git a/apps/sim/lib/mothership/tools/client/terminal-tool-execution.ts b/apps/sim/lib/mothership/tools/client/terminal-tool-execution.ts index 8cb295c85ca..a7bddd20de2 100644 --- a/apps/sim/lib/mothership/tools/client/terminal-tool-execution.ts +++ b/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)