Add piping

This commit is contained in:
Siddharth Ganesan
2026-03-10 12:36:28 -07:00
parent 9ca539b626
commit 6db8bc4934
9 changed files with 288 additions and 3 deletions
@@ -610,6 +610,7 @@ export async function POST(req: NextRequest) {
workflowVariables = {},
workflowId,
isCustomTool = false,
_sandboxFiles,
} = body
const executionParams = { ...params }
@@ -722,6 +723,7 @@ export async function POST(req: NextRequest) {
code: codeForE2B,
language: CodeLanguage.JavaScript,
timeoutMs: timeout,
sandboxFiles: _sandboxFiles,
})
const executionTime = Date.now() - execStart
stdout += e2bStdout
@@ -785,6 +787,7 @@ export async function POST(req: NextRequest) {
code: codeForE2B,
language: CodeLanguage.Python,
timeoutMs: timeout,
sandboxFiles: _sandboxFiles,
})
const executionTime = Date.now() - execStart
stdout += e2bStdout
@@ -26,6 +26,7 @@ import type {
import { SUBAGENT_LABELS } from '../types'
import {
extractFileResource,
extractFunctionExecuteResource,
extractResourcesFromHistory,
extractTableResource,
extractWorkflowResource,
@@ -361,6 +362,22 @@ export function useChat(workspaceId: string, initialChatId?: string): UseChatRet
queryKey: workspaceFilesKeys.content(workspaceId, resource.id),
})
}
} else if (toolName === 'function_execute') {
resource = extractFunctionExecuteResource(parsed, storedArgs)
if (resource) {
if (resource.type === 'table') {
lastTableId = resource.id
queryClient.invalidateQueries({ queryKey: tableKeys.detail(resource.id) })
queryClient.invalidateQueries({ queryKey: tableKeys.rowsRoot(resource.id) })
} else if (resource.type === 'file') {
queryClient.invalidateQueries({
queryKey: workspaceFilesKeys.list(workspaceId),
})
queryClient.invalidateQueries({
queryKey: workspaceFilesKeys.content(workspaceId, resource.id),
})
}
}
} else if (toolName === 'create_workflow' || toolName === 'edit_workflow') {
resource = extractWorkflowResource(parsed, lastWorkflowId)
if (resource) {
@@ -6,6 +6,7 @@ export const RESOURCE_TOOL_NAMES = new Set([
'workspace_file',
'create_workflow',
'edit_workflow',
'function_execute',
])
function getResultData(parsed: SSEPayload): Record<string, unknown> | undefined {
@@ -60,6 +61,34 @@ export function extractFileResource(
return null
}
export function extractFunctionExecuteResource(
parsed: SSEPayload,
storedArgs: Record<string, unknown> | undefined
): MothershipResource | null {
const topResult = (parsed.result ??
(typeof parsed.data === 'object' ? parsed.data?.result : undefined)) as
| Record<string, unknown>
| undefined
if (topResult?.tableId) {
return {
type: 'table',
id: topResult.tableId as string,
title: (topResult.tableName as string) || 'Table',
}
}
if (topResult?.fileId) {
return {
type: 'file',
id: topResult.fileId as string,
title: (topResult.fileName as string) || 'File',
}
}
return null
}
export function extractWorkflowResource(
parsed: SSEPayload,
fallbackWorkflowId: string | null
@@ -113,6 +142,9 @@ export function extractResourcesFromHistory(messages: TaskStoredMessage[]): Moth
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 === 'create_workflow' || tc.name === 'edit_workflow') {
resource = extractWorkflowResource(payload, lastWorkflowId)
if (resource) lastWorkflowId = resource.id
@@ -1,4 +1,7 @@
import { db } from '@sim/db'
import { userTableRows } from '@sim/db/schema'
import { createLogger } from '@sim/logger'
import { eq } from 'drizzle-orm'
import {
TOOL_DECISION_INITIAL_POLL_MS,
TOOL_DECISION_MAX_POLL_MS,
@@ -18,6 +21,7 @@ import type {
StreamingContext,
ToolCallResult,
} from '@/lib/copilot/orchestrator/types'
import { getTableById } from '@/lib/table/service'
import { uploadWorkspaceFile } from '@/lib/uploads/contexts/workspace/workspace-file-manager'
const logger = createLogger('CopilotSseToolExecution')
@@ -94,6 +98,114 @@ async function maybeWriteOutputToFile(
}
}
const MAX_OUTPUT_TABLE_ROWS = 10_000
const BATCH_CHUNK_SIZE = 500
async function maybeWriteOutputToTable(
toolName: string,
params: Record<string, unknown> | undefined,
result: ToolCallResult,
context: ExecutionContext
): Promise<ToolCallResult> {
if (toolName !== 'function_execute') 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 rawOutput = result.output
let rows: Array<Record<string, unknown>>
if (rawOutput && typeof rawOutput === 'object' && 'result' in rawOutput) {
const inner = (rawOutput as Record<string, unknown>).result
if (Array.isArray(inner)) {
rows = inner
} else {
return {
success: false,
error: 'outputTable requires the code to return an array of objects',
}
}
} else if (Array.isArray(rawOutput)) {
rows = rawOutput
} else {
return {
success: false,
error: 'outputTable requires the code to return an array of objects',
}
}
if (rows.length > MAX_OUTPUT_TABLE_ROWS) {
return {
success: false,
error: `outputTable row limit exceeded: got ${rows.length}, max is ${MAX_OUTPUT_TABLE_ROWS}`,
}
}
if (rows.length === 0) {
return {
success: false,
error: 'outputTable requires at least one row — code returned an empty array',
}
}
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('Tool output written to table', {
toolName,
tableId: outputTable,
rowCount: rows.length,
})
return {
success: true,
output: {
message: `Wrote ${rows.length} rows to table ${outputTable}`,
tableId: outputTable,
rowCount: rows.length,
},
}
} catch (err) {
logger.warn('Failed to write tool output to table', {
toolName,
outputTable,
error: err instanceof Error ? err.message : String(err),
})
return {
success: false,
error: `Failed to write to table: ${err instanceof Error ? err.message : String(err)}`,
}
}
}
export async function executeToolAndReport(
toolCallId: string,
context: StreamingContext,
@@ -117,6 +229,7 @@ export async function executeToolAndReport(
try {
let result = await executeToolServerSide(toolCall, execContext)
result = await maybeWriteOutputToFile(toolCall.name, toolCall.params, result, execContext)
result = await maybeWriteOutputToTable(toolCall.name, toolCall.params, result, execContext)
toolCall.status = result.success ? 'success' : 'error'
toolCall.result = result
toolCall.error = result.error
@@ -12,6 +12,11 @@ import { generateRequestId } from '@/lib/core/utils/request'
import { getCredentialActorContext } from '@/lib/credentials/access'
import { getAccessibleOAuthCredentials } from '@/lib/credentials/environment'
import { getEffectiveDecryptedEnv } from '@/lib/environment/utils'
import { getTableById, queryRows } from '@/lib/table/service'
import {
downloadWorkspaceFile,
listWorkspaceFiles,
} from '@/lib/uploads/contexts/workspace/workspace-file-manager'
import { getWorkflowById } from '@/lib/workflows/utils'
import { refreshTokenIfNeeded } from '@/app/api/auth/oauth/utils'
import { resolveEnvVarReferences } from '@/executor/utils/reference-validation'
@@ -143,6 +148,97 @@ export async function executeIntegrationToolDirect(
executionParams.blockNameMapping = {}
executionParams.language = executionParams.language || 'javascript'
executionParams.timeout = executionParams.timeout || 30000
if (isHosted && workspaceId) {
const sandboxFiles: Array<{ path: string; content: string }> = []
const MAX_FILE_SIZE = 10 * 1024 * 1024
const MAX_TOTAL_SIZE = 50 * 1024 * 1024
const TEXT_EXTENSIONS = new Set([
'csv',
'json',
'txt',
'md',
'html',
'xml',
'tsv',
'yaml',
'yml',
])
let totalSize = 0
const inputFilePaths = executionParams.inputFiles as string[] | undefined
if (inputFilePaths?.length) {
const allFiles = await listWorkspaceFiles(workspaceId)
for (const filePath of inputFilePaths) {
const fileName = filePath.replace(/^files\//, '')
const ext = fileName.split('.').pop()?.toLowerCase() ?? ''
if (!TEXT_EXTENSIONS.has(ext)) {
logger.warn('Skipping non-text sandbox input file', { fileName, ext })
continue
}
const record = allFiles.find(
(f) => f.name === fileName || f.name.normalize('NFC') === fileName.normalize('NFC')
)
if (!record) {
logger.warn('Sandbox input file not found', { fileName })
continue
}
if (record.size > MAX_FILE_SIZE) {
logger.warn('Sandbox input file exceeds size limit', { fileName, size: record.size })
continue
}
if (totalSize + record.size > MAX_TOTAL_SIZE) {
logger.warn('Sandbox input total size limit reached, skipping remaining files')
break
}
const buffer = await downloadWorkspaceFile(record)
totalSize += buffer.length
sandboxFiles.push({ path: `/home/user/${fileName}`, content: buffer.toString('utf-8') })
}
}
const inputTableIds = executionParams.inputTables as string[] | undefined
if (inputTableIds?.length) {
for (const tableId of inputTableIds) {
const table = await getTableById(tableId)
if (!table) {
logger.warn('Sandbox input table not found', { tableId })
continue
}
const { rows } = await queryRows(tableId, workspaceId, { limit: 10000 }, 'sandbox-input')
const cols = (table.schema as { columns: Array<{ name: string }> }).columns.map(
(c) => c.name
)
const csvLines = [cols.join(',')]
for (const row of rows) {
csvLines.push(
cols
.map((c) => JSON.stringify((row.data as Record<string, unknown>)[c] ?? ''))
.join(',')
)
}
const csvContent = csvLines.join('\n')
if (totalSize + csvContent.length > MAX_TOTAL_SIZE) {
logger.warn('Sandbox input total size limit reached, skipping remaining tables')
break
}
totalSize += csvContent.length
sandboxFiles.push({ path: `/home/user/tables/${tableId}.csv`, content: csvContent })
}
}
if (sandboxFiles.length > 0) {
executionParams._sandboxFiles = sandboxFiles
logger.info('Prepared sandbox input files', {
fileCount: sandboxFiles.length,
totalSize,
paths: sandboxFiles.map((f) => f.path),
})
}
executionParams.inputFiles = undefined
executionParams.inputTables = undefined
}
}
const result = await executeTool(toolName, executionParams)
+2 -2
View File
@@ -22,8 +22,8 @@ export const isTest = env.NODE_ENV === 'test'
* Is this the hosted version of the application
*/
export const isHosted =
getEnv('NEXT_PUBLIC_APP_URL') === 'https://www.sim.ai' ||
getEnv('NEXT_PUBLIC_APP_URL') === 'https://www.staging.sim.ai'
getEnv('NEXT_PUBLIC_APP_URL') === 'https://www.sim.ai' ||
getEnv('NEXT_PUBLIC_APP_URL') === 'https://www.staging.sim.ai'
/**
* Is billing enforcement enabled
+17
View File
@@ -3,10 +3,16 @@ import { createLogger } from '@sim/logger'
import { env } from '@/lib/core/config/env'
import { CodeLanguage } from '@/lib/execution/languages'
export interface SandboxFile {
path: string
content: string
}
export interface E2BExecutionRequest {
code: string
language: CodeLanguage
timeoutMs: number
sandboxFiles?: SandboxFile[]
}
export interface E2BExecutionResult {
@@ -29,6 +35,17 @@ export async function executeInE2B(req: E2BExecutionRequest): Promise<E2BExecuti
const sandbox = await Sandbox.create({ apiKey })
const sandboxId = sandbox.sandboxId
if (req.sandboxFiles?.length) {
for (const file of req.sandboxFiles) {
await sandbox.files.write(file.path, file.content)
}
logger.info('Wrote sandbox input files', {
sandboxId,
fileCount: req.sandboxFiles.length,
paths: req.sandboxFiles.map((f) => f.path),
})
}
const stdoutChunks = []
try {
+7 -1
View File
@@ -80,7 +80,7 @@ export const functionExecuteTool: ToolConfig<CodeExecutionInput, CodeExecutionOu
? params.code.map((c: { content: string }) => c.content).join('\n')
: params.code
return {
const body: Record<string, unknown> = {
code: codeContent,
language: params.language || DEFAULT_CODE_LANGUAGE,
timeout: params.timeout || DEFAULT_EXECUTION_TIMEOUT_MS,
@@ -93,6 +93,12 @@ export const functionExecuteTool: ToolConfig<CodeExecutionInput, CodeExecutionOu
userId: params._context?.userId,
isCustomTool: params.isCustomTool || false,
}
if (params._sandboxFiles) {
body._sandboxFiles = params._sandboxFiles
}
return body
},
},
+1
View File
@@ -17,6 +17,7 @@ export interface CodeExecutionInput {
userId?: string
}
isCustomTool?: boolean
_sandboxFiles?: Array<{ path: string; content: string }>
}
export interface CodeExecutionOutput extends ToolResponse {