From f448134618beee0ea026edc59ac88cb4d43217b7 Mon Sep 17 00:00:00 2001 From: Siddharth Ganesan Date: Tue, 10 Mar 2026 20:34:58 -0700 Subject: [PATCH] Subagent tool call persistence --- apps/sim/app/api/mothership/chat/route.ts | 1 + .../[workspaceId]/home/hooks/use-chat.ts | 29 ++++- .../app/workspace/[workspaceId]/home/utils.ts | 13 +- apps/sim/hooks/queries/tasks.ts | 1 + .../orchestrator/sse/handlers/handlers.ts | 6 + .../sse/handlers/tool-execution.ts | 114 ++++++++++++++++++ .../lib/copilot/orchestrator/stream/core.ts | 8 ++ apps/sim/lib/copilot/orchestrator/types.ts | 3 +- apps/sim/lib/file-parsers/csv-parser.ts | 6 +- 9 files changed, 170 insertions(+), 11 deletions(-) diff --git a/apps/sim/app/api/mothership/chat/route.ts b/apps/sim/app/api/mothership/chat/route.ts index 38c48ee527..3f505d3484 100644 --- a/apps/sim/app/api/mothership/chat/route.ts +++ b/apps/sim/app/api/mothership/chat/route.ts @@ -206,6 +206,7 @@ export async function POST(req: NextRequest) { : 'error' : block.toolCall.status, result: block.toolCall.result, + ...(block.calledBy ? { calledBy: block.calledBy } : {}), } } return stored 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 375c1778d9..91e146ade4 100644 --- a/apps/sim/app/workspace/[workspaceId]/home/hooks/use-chat.ts +++ b/apps/sim/app/workspace/[workspaceId]/home/hooks/use-chat.ts @@ -70,6 +70,7 @@ function mapStoredBlock(block: TaskStoredContentBlock): ContentBlock { name: block.toolCall.name ?? 'unknown', status: STATE_TO_STATUS[block.toolCall.state ?? ''] ?? 'success', displayTitle: block.toolCall.display?.text, + calledBy: block.toolCall.calledBy, result: block.toolCall.result, } } @@ -383,11 +384,22 @@ export function useChat(workspaceId: string, initialChatId?: string): UseChatRet let resource: MothershipResource | null = null if (toolName === 'user_table') { - resource = extractTableResource(parsed, storedArgs, lastTableId) - if (resource) { - lastTableId = resource.id - queryClient.invalidateQueries({ queryKey: tableKeys.detail(resource.id) }) - queryClient.invalidateQueries({ queryKey: tableKeys.rowsRoot(resource.id) }) + const redirected = extractFunctionExecuteResource(parsed, storedArgs) + if (redirected?.type === 'file') { + resource = redirected + queryClient.invalidateQueries({ + queryKey: workspaceFilesKeys.list(workspaceId), + }) + queryClient.invalidateQueries({ + queryKey: workspaceFilesKeys.content(workspaceId, resource.id), + }) + } else { + resource = extractTableResource(parsed, storedArgs, lastTableId) + if (resource) { + lastTableId = resource.id + queryClient.invalidateQueries({ queryKey: tableKeys.detail(resource.id) }) + queryClient.invalidateQueries({ queryKey: tableKeys.rowsRoot(resource.id) }) + } } } else if (toolName === 'workspace_file') { resource = extractFileResource(parsed, storedArgs) @@ -415,6 +427,13 @@ export function useChat(workspaceId: string, initialChatId?: string): UseChatRet }) } } + } else if (toolName === 'read') { + resource = extractFunctionExecuteResource(parsed, storedArgs) + if (resource?.type === 'table') { + lastTableId = resource.id + queryClient.invalidateQueries({ queryKey: tableKeys.detail(resource.id) }) + queryClient.invalidateQueries({ queryKey: tableKeys.rowsRoot(resource.id) }) + } } else if (toolName === 'create_workflow' || toolName === 'edit_workflow') { resource = extractWorkflowResource(parsed, lastWorkflowId) if (resource) { diff --git a/apps/sim/app/workspace/[workspaceId]/home/utils.ts b/apps/sim/app/workspace/[workspaceId]/home/utils.ts index 728eb3619a..28697308a7 100644 --- a/apps/sim/app/workspace/[workspaceId]/home/utils.ts +++ b/apps/sim/app/workspace/[workspaceId]/home/utils.ts @@ -7,6 +7,7 @@ export const RESOURCE_TOOL_NAMES = new Set([ 'create_workflow', 'edit_workflow', 'function_execute', + 'read', ]) function getResultData(parsed: SSEPayload): Record | undefined { @@ -138,13 +139,21 @@ export function extractResourcesFromHistory(messages: TaskStoredMessage[]): Moth let resource: MothershipResource | null = null if (tc.name === 'user_table') { - resource = extractTableResource(payload, args, lastTableId) - if (resource) lastTableId = resource.id + const redirected = extractFunctionExecuteResource(payload, args) + if (redirected?.type === 'file') { + resource = redirected + } else { + resource = extractTableResource(payload, args, lastTableId) + if (resource) lastTableId = resource.id + } } else if (tc.name === 'workspace_file') { resource = extractFileResource(payload, args) } else if (tc.name === 'function_execute') { resource = extractFunctionExecuteResource(payload, args) if (resource?.type === 'table') lastTableId = resource.id + } else if (tc.name === 'read') { + resource = extractFunctionExecuteResource(payload, args) + if (resource?.type === 'table') lastTableId = resource.id } else if (tc.name === 'create_workflow' || tc.name === 'edit_workflow') { resource = extractWorkflowResource(payload, lastWorkflowId) if (resource) lastWorkflowId = resource.id diff --git a/apps/sim/hooks/queries/tasks.ts b/apps/sim/hooks/queries/tasks.ts index b3f5b2e9ab..f19d977bf7 100644 --- a/apps/sim/hooks/queries/tasks.ts +++ b/apps/sim/hooks/queries/tasks.ts @@ -41,6 +41,7 @@ export interface TaskStoredContentBlock { params?: Record result?: { success: boolean; output?: unknown; error?: string } display?: { text?: string } + calledBy?: string } | null } diff --git a/apps/sim/lib/copilot/orchestrator/sse/handlers/handlers.ts b/apps/sim/lib/copilot/orchestrator/sse/handlers/handlers.ts index 47d6e48857..1281293e19 100644 --- a/apps/sim/lib/copilot/orchestrator/sse/handlers/handlers.ts +++ b/apps/sim/lib/copilot/orchestrator/sse/handlers/handlers.ts @@ -484,6 +484,12 @@ export const subAgentHandlers: Record = { } if (!context.toolCalls.has(toolCallId)) { context.toolCalls.set(toolCallId, toolCall) + const parentToolCall = context.toolCalls.get(parentToolCallId) + addContentBlock(context, { + type: 'tool_call', + toolCall, + calledBy: parentToolCall?.name, + }) } if (isPartial) return diff --git a/apps/sim/lib/copilot/orchestrator/sse/handlers/tool-execution.ts b/apps/sim/lib/copilot/orchestrator/sse/handlers/tool-execution.ts index 1720f5f89f..3effbb3488 100644 --- a/apps/sim/lib/copilot/orchestrator/sse/handlers/tool-execution.ts +++ b/apps/sim/lib/copilot/orchestrator/sse/handlers/tool-execution.ts @@ -206,6 +206,119 @@ async function maybeWriteOutputToTable( } } +async function maybeWriteReadCsvToTable( + toolName: string, + params: Record | undefined, + result: ToolCallResult, + context: ExecutionContext +): Promise { + if (toolName !== 'read') return result + if (!result.success || !result.output) return result + if (!context.workspaceId || !context.userId) return result + + const outputTable = params?.outputTable as string | undefined + if (!outputTable) return result + + try { + const table = await getTableById(outputTable) + if (!table) { + return { success: false, error: `Table "${outputTable}" not found` } + } + + const output = result.output as Record + const content = (output.content as string) || '' + if (!content.trim()) { + return { success: false, error: 'File has no content to import into table' } + } + + const filePath = (params?.path as string) || '' + const ext = filePath.split('.').pop()?.toLowerCase() + + let rows: Record[] + + if (ext === 'json') { + const parsed = JSON.parse(content) + if (!Array.isArray(parsed)) { + return { + success: false, + error: 'JSON file must contain an array of objects for table import', + } + } + rows = parsed + } else { + const { parse } = await import('csv-parse/sync') + rows = parse(content, { + columns: true, + skip_empty_lines: true, + trim: true, + relax_column_count: true, + relax_quotes: true, + skip_records_with_error: true, + cast: false, + }) as Record[] + } + + if (rows.length === 0) { + return { success: false, error: 'File has no data rows to import' } + } + + if (rows.length > MAX_OUTPUT_TABLE_ROWS) { + return { + success: false, + error: `Row limit exceeded: got ${rows.length}, max is ${MAX_OUTPUT_TABLE_ROWS}`, + } + } + + await db.transaction(async (tx) => { + await tx.delete(userTableRows).where(eq(userTableRows.tableId, outputTable)) + + const now = new Date() + for (let i = 0; i < rows.length; i += BATCH_CHUNK_SIZE) { + const chunk = rows.slice(i, i + BATCH_CHUNK_SIZE) + const values = chunk.map((rowData, j) => ({ + id: `row_${crypto.randomUUID().replace(/-/g, '')}`, + tableId: outputTable, + workspaceId: context.workspaceId!, + data: rowData, + position: i + j, + createdAt: now, + updatedAt: now, + createdBy: context.userId, + })) + await tx.insert(userTableRows).values(values) + } + }) + + logger.info('Read output written to table', { + toolName, + tableId: outputTable, + tableName: table.name, + rowCount: rows.length, + filePath, + }) + + return { + success: true, + output: { + message: `Imported ${rows.length} rows from "${filePath}" into table "${table.name}"`, + tableId: outputTable, + tableName: table.name, + rowCount: rows.length, + }, + } + } catch (err) { + logger.warn('Failed to write read output to table', { + toolName, + outputTable, + error: err instanceof Error ? err.message : String(err), + }) + return { + success: false, + error: `Failed to import into table: ${err instanceof Error ? err.message : String(err)}`, + } + } +} + export async function executeToolAndReport( toolCallId: string, context: StreamingContext, @@ -230,6 +343,7 @@ export async function executeToolAndReport( let result = await executeToolServerSide(toolCall, execContext) result = await maybeWriteOutputToFile(toolCall.name, toolCall.params, result, execContext) result = await maybeWriteOutputToTable(toolCall.name, toolCall.params, result, execContext) + result = await maybeWriteReadCsvToTable(toolCall.name, toolCall.params, result, execContext) toolCall.status = result.success ? 'success' : 'error' toolCall.result = result toolCall.error = result.error diff --git a/apps/sim/lib/copilot/orchestrator/stream/core.ts b/apps/sim/lib/copilot/orchestrator/stream/core.ts index 6c4924f7d8..4083c293e9 100644 --- a/apps/sim/lib/copilot/orchestrator/stream/core.ts +++ b/apps/sim/lib/copilot/orchestrator/stream/core.ts @@ -176,12 +176,20 @@ export async function runStreamLoop( if (normalizedEvent.type === 'subagent_start') { const eventData = normalizedEvent.data as Record | undefined const toolCallId = eventData?.tool_call_id as string | undefined + const subagentName = normalizedEvent.subagent || (eventData?.agent as string | undefined) if (toolCallId) { context.subAgentParentStack.push(toolCallId) context.subAgentParentToolCallId = toolCallId context.subAgentContent[toolCallId] = '' context.subAgentToolCalls[toolCallId] = [] } + if (subagentName) { + context.contentBlocks.push({ + type: 'subagent', + content: subagentName, + timestamp: Date.now(), + }) + } continue } diff --git a/apps/sim/lib/copilot/orchestrator/types.ts b/apps/sim/lib/copilot/orchestrator/types.ts index 483ecffb0f..6f5d57350f 100644 --- a/apps/sim/lib/copilot/orchestrator/types.ts +++ b/apps/sim/lib/copilot/orchestrator/types.ts @@ -59,12 +59,13 @@ export interface ToolCallResult { error?: string } -export type ContentBlockType = 'text' | 'thinking' | 'tool_call' | 'subagent_text' +export type ContentBlockType = 'text' | 'thinking' | 'tool_call' | 'subagent_text' | 'subagent' export interface ContentBlock { type: ContentBlockType content?: string toolCall?: ToolCallState + calledBy?: string timestamp: number } diff --git a/apps/sim/lib/file-parsers/csv-parser.ts b/apps/sim/lib/file-parsers/csv-parser.ts index ad9bf74b86..cabaa6add2 100644 --- a/apps/sim/lib/file-parsers/csv-parser.ts +++ b/apps/sim/lib/file-parsers/csv-parser.ts @@ -37,9 +37,9 @@ export class CsvParser implements FileParser { `Parsing CSV buffer, size: ${bufferSize} bytes (${(bufferSize / 1024 / 1024).toFixed(2)} MB)` ) - const stream = Readable.from(buffer, { - highWaterMark: CONFIG.STREAM_CHUNK_SIZE, - }) + const stream = new Readable({ read() {} }) + stream.push(buffer) + stream.push(null) return this.parseStream(stream) }