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..1b3788424c0 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,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 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..021a7599f2c --- /dev/null +++ b/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 + +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 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 | 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 +} 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..e7e285f64bc --- /dev/null +++ b/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() + +/** 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 { + 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 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 => { + // 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 => { + // 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) +} 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..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 @@ -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,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 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) + 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 15de9ecf2f0..bc91bbc22d4 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, parseStreamBatchResponse, resolveChatIdFromStreamBatch, + STREAM_BATCH_FETCH_TIMEOUT_MS, + STREAM_IDLE_TIMEOUT_MS, type StreamBatchResponse, StreamGoneError, } from './stream-protocol' @@ -286,9 +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 -/** 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 +910,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 +966,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 +1048,7 @@ export function useChat( const resetHomeChatState = useCallback(() => { const abandonedDesktopScopeId = desktopScopeIdRef.current + detachLiveTurnClientToolsRef.current() cancelActiveStreamRecovery() streamGenRef.current++ cancelActiveStreamReader() @@ -1530,61 +1536,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 +1607,38 @@ 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. 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 + if ( + !isDesktopApp() || + requestModeRef.current === 'assistant' || + !sendingRef.current || + !streamId || + pendingChatAdmissionRef.current + ) { + return false + } + detachClientTools({ + streamId, + chatId: chatIdRef.current, + afterCursor: lastCursorRef.current, + traceparent: streamTraceparentRef.current, + workspaceId, + scopeKey, + }) + return true + } + const recoverPendingClientWorkflowTools = useCallback( async (nextMessages: ChatMessage[]) => { const pending: ToolCallInfo[] = [] @@ -1767,6 +1757,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() @@ -2218,6 +2209,7 @@ export function useChat( return { sawStreamError: false, sawComplete: false } } streamReaderRef.current = reader + if (streamIdRef.current) reattachClientTools(streamIdRef.current) try { await readSSELines(reader, { @@ -2247,7 +2239,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 +2358,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 +2547,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) } @@ -2575,6 +2561,7 @@ export function useChat( } if (isStaleReconnect()) { + await sseRes.body.cancel() return { error: false, aborted: true } } @@ -4892,11 +4879,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..f89492bc82d --- /dev/null +++ b/apps/sim/lib/mothership/tools/client/launch-local-filesystem-tool.ts @@ -0,0 +1,57 @@ +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') + +/** + * 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 + } + + 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..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' @@ -38,7 +39,7 @@ const VFS_GLOB_OPTIONS: micromatch.Options = { noext: true, } -interface LocalFilesystemExecutionContext { +export interface LocalFilesystemExecutionContext { workspaceId?: string chatId?: string signal?: AbortSignal @@ -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)