improvement(mothership): table speed parity — keyset reads, limit bounds, background import/delete jobs

Mothership's table operations lagged the fast paths the tables feature
already has:

- function_execute mounted inputTables via queryRows with defaults,
  silently truncating the sandbox CSV to 100 rows and paying for a count
  and execution metadata it never used. Mounts now drain the keyset
  export reader page by page, remap stored column-id keys to display
  names (headers previously didn't match id-keyed cell data), mount
  tables in parallel, and count toward the 50MB sandbox mount budget.
- user_table query_rows accepted unbounded limits (whole table into one
  tool result) and only offset paging. It now enforces the contracts'
  MAX_QUERY_LIMIT, skips execution metadata, accepts the keyset `after`
  cursor, and returns `nextCursor` when a default-order page fills.
- import_file / create_from_file / delete_rows_by_filter mutated without
  claiming the per-table job slot, racing running background jobs, and
  ran whole imports inline in the chat request. CSV/TSV imports ≥8MB and
  unbounded filter-deletes matching >1000 rows now dispatch the same
  trigger.dev jobs the UI routes use (slot claim, release-on-dispatch-
  failure, delete mask); inline imports claim and release the slot. A new
  get_job operation surfaces the table's derived job state for polling.
- TableImportPayload grows deleteSourceFile (default true) so Mothership
  imports of persistent workspace files survive the worker's single-use
  source cleanup.

Generated tool catalog artifacts regenerated from simstudioai/copilot#289.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: TheodoreSpeaks

apps/sim/lib/copilot/generated/tool-catalog-v1.ts

@@ -3868,6 +3868,22 @@ export const UserTable: ToolCatalogEntry = {
         type: 'object',
         description: 'Arguments for the operation',
         properties: {
+          after: {
+            type: 'object',
+            description:
+              "Keyset cursor for query_rows: pass the nextCursor object ({ orderKey, id }) from the previous page's response to fetch the next page on the default row order. Cannot be combined with sort; takes precedence over offset.",
+            properties: {
+              id: {
+                type: 'string',
+                description: 'id of the last row of the previous page (from nextCursor).',
+              },
+              orderKey: {
+                type: 'string',
+                description: 'orderKey of the last row of the previous page (from nextCursor).',
+              },
+            },
+            required: ['orderKey', 'id'],
+          },
           autoRun: {
             type: 'boolean',
             description:
@@ -3956,7 +3972,8 @@ export const UserTable: ToolCatalogEntry = {
           },
           limit: {
             type: 'number',
-            description: 'Maximum rows to return or affect (optional, default 100)',
+            description:
+              'Maximum rows to return or affect (optional; default 100, max 1000). For delete_rows_by_filter, omitting it lets matches above 1000 run as a background job.',
           },
           mapping: {
             type: 'object',
@@ -4009,7 +4026,8 @@ export const UserTable: ToolCatalogEntry = {
           },
           offset: {
             type: 'number',
-            description: 'Number of rows to skip (optional for query_rows, default 0)',
+            description:
+              'Number of rows to skip (optional for query_rows, default 0). For paging past more than a few pages, prefer the after cursor — offset re-scans every prior row.',
           },
           outputColumnNames: {
             type: 'object',
@@ -4146,6 +4164,7 @@ export const UserTable: ToolCatalogEntry = {
           'import_file',
           'get',
           'get_schema',
+          'get_job',
           'delete',
           'rename',
           'insert_row',
@@ -4524,6 +4543,7 @@ export const UserTableOperation = {
   importFile: 'import_file',
   get: 'get',
   getSchema: 'get_schema',
+  getJob: 'get_job',
   delete: 'delete',
   rename: 'rename',
   insertRow: 'insert_row',
@@ -4560,6 +4580,7 @@ export const UserTableOperationValues = [
   UserTableOperation.importFile,
   UserTableOperation.get,
   UserTableOperation.getSchema,
+  UserTableOperation.getJob,
   UserTableOperation.delete,
   UserTableOperation.rename,
   UserTableOperation.insertRow,

apps/sim/lib/copilot/generated/tool-schemas-v1.ts

@@ -3598,6 +3598,22 @@ export const TOOL_RUNTIME_SCHEMAS: Record<string, ToolRuntimeSchemaEntry> = {
           type: 'object',
           description: 'Arguments for the operation',
           properties: {
+            after: {
+              type: 'object',
+              description:
+                "Keyset cursor for query_rows: pass the nextCursor object ({ orderKey, id }) from the previous page's response to fetch the next page on the default row order. Cannot be combined with sort; takes precedence over offset.",
+              properties: {
+                id: {
+                  type: 'string',
+                  description: 'id of the last row of the previous page (from nextCursor).',
+                },
+                orderKey: {
+                  type: 'string',
+                  description: 'orderKey of the last row of the previous page (from nextCursor).',
+                },
+              },
+              required: ['orderKey', 'id'],
+            },
             autoRun: {
               type: 'boolean',
               description:
@@ -3694,7 +3710,8 @@ export const TOOL_RUNTIME_SCHEMAS: Record<string, ToolRuntimeSchemaEntry> = {
             },
             limit: {
               type: 'number',
-              description: 'Maximum rows to return or affect (optional, default 100)',
+              description:
+                'Maximum rows to return or affect (optional; default 100, max 1000). For delete_rows_by_filter, omitting it lets matches above 1000 run as a background job.',
             },
             mapping: {
               type: 'object',
@@ -3753,7 +3770,8 @@ export const TOOL_RUNTIME_SCHEMAS: Record<string, ToolRuntimeSchemaEntry> = {
             },
             offset: {
               type: 'number',
-              description: 'Number of rows to skip (optional for query_rows, default 0)',
+              description:
+                'Number of rows to skip (optional for query_rows, default 0). For paging past more than a few pages, prefer the after cursor — offset re-scans every prior row.',
             },
             outputColumnNames: {
               type: 'object',
@@ -3902,6 +3920,7 @@ export const TOOL_RUNTIME_SCHEMAS: Record<string, ToolRuntimeSchemaEntry> = {
             'import_file',
             'get',
             'get_schema',
+            'get_job',
             'delete',
             'rename',
             'insert_row',

apps/sim/lib/copilot/tools/handlers/function-execute.test.ts

@@ -0,0 +1,135 @@
+/**
+ * @vitest-environment node
+ */
+
+import { beforeEach, describe, expect, it, vi } from 'vitest'
+
+const { mockGetTableById, mockSelectExportRowPage, mockExecuteTool } = vi.hoisted(() => ({
+  mockGetTableById: vi.fn(),
+  mockSelectExportRowPage: vi.fn(),
+  mockExecuteTool: vi.fn(),
+}))
+
+vi.mock('@/lib/table/service', () => ({
+  getTableById: mockGetTableById,
+  listTables: vi.fn(),
+  selectExportRowPage: mockSelectExportRowPage,
+}))
+
+vi.mock('@/tools', () => ({
+  executeTool: mockExecuteTool,
+}))
+
+vi.mock('@/lib/uploads/contexts/workspace/workspace-file-manager', () => ({
+  fetchWorkspaceFileBuffer: vi.fn(),
+  findWorkspaceFileRecord: vi.fn(),
+  getSandboxWorkspaceFilePath: vi.fn(),
+  listWorkspaceFiles: vi.fn(),
+}))
+
+vi.mock('@/lib/uploads/contexts/workspace/workspace-file-folder-manager', () => ({
+  listWorkspaceFileFolders: vi.fn(),
+}))
+
+vi.mock('@/lib/copilot/vfs/workflow-alias-resolver', () => ({
+  resolveWorkflowAliasForWorkspace: vi.fn(),
+}))
+
+import { executeFunctionExecute } from '@/lib/copilot/tools/handlers/function-execute'
+
+const PAGE_SIZE = 5000
+
+function buildTable() {
+  return {
+    id: 'tbl_1',
+    name: 'People',
+    description: null,
+    schema: {
+      columns: [
+        { id: 'col_name', name: 'name', type: 'string' },
+        { id: 'col_age', name: 'age', type: 'number' },
+      ],
+    },
+    metadata: null,
+    rowCount: PAGE_SIZE + 2,
+    maxRows: 100000,
+    workspaceId: 'workspace-1',
+    createdBy: 'user-1',
+    archivedAt: null,
+    createdAt: new Date('2024-01-01'),
+    updatedAt: new Date('2024-01-01'),
+  }
+}
+
+function makeRows(start: number, count: number) {
+  return Array.from({ length: count }, (_, i) => ({
+    id: `row_${start + i}`,
+    data: { col_name: `person ${start + i}`, col_age: start + i },
+    position: start + i,
+  }))
+}
+
+describe('executeFunctionExecute table mounts', () => {
+  beforeEach(() => {
+    vi.clearAllMocks()
+    mockGetTableById.mockResolvedValue(buildTable())
+    mockExecuteTool.mockResolvedValue({ success: true, output: {} })
+  })
+
+  it('drains every row through the keyset export reader, not a single 100-row page', async () => {
+    mockSelectExportRowPage
+      .mockResolvedValueOnce(makeRows(0, PAGE_SIZE))
+      .mockResolvedValueOnce(makeRows(PAGE_SIZE, 2))
+
+    await executeFunctionExecute(
+      { code: 'print(1)', inputTables: ['tbl_1'] },
+      { userId: 'user-1', workflowId: 'wf-1', workspaceId: 'workspace-1' }
+    )
+
+    expect(mockSelectExportRowPage).toHaveBeenCalledTimes(2)
+    expect(mockSelectExportRowPage.mock.calls[0][1]).toBeNull()
+    expect(mockSelectExportRowPage.mock.calls[1][1]).toEqual({
+      position: PAGE_SIZE - 1,
+      id: `row_${PAGE_SIZE - 1}`,
+    })
+
+    const params = mockExecuteTool.mock.calls[0][1] as {
+      _sandboxFiles: Array<{ path: string; content: string }>
+    }
+    const mount = params._sandboxFiles.find((f) => f.path === '/home/user/tables/tbl_1.csv')
+    expect(mount).toBeDefined()
+    const lines = mount!.content.split('\n')
+    expect(lines[0]).toBe('name,age')
+    expect(lines).toHaveLength(1 + PAGE_SIZE + 2)
+    expect(lines[1]).toBe('person 0,0')
+    expect(lines[lines.length - 1]).toBe(`person ${PAGE_SIZE + 1},${PAGE_SIZE + 1}`)
+  })
+
+  it('maps stored column-id keys back to display-name headers', async () => {
+    mockSelectExportRowPage.mockResolvedValueOnce([
+      { id: 'row_1', data: { col_name: 'Alice', col_age: 30 }, position: 0 },
+    ])
+
+    await executeFunctionExecute(
+      { code: 'print(1)', inputTables: ['tbl_1'] },
+      { userId: 'user-1', workflowId: 'wf-1', workspaceId: 'workspace-1' }
+    )
+
+    const params = mockExecuteTool.mock.calls[0][1] as {
+      _sandboxFiles: Array<{ path: string; content: string }>
+    }
+    expect(params._sandboxFiles[0].content).toBe('name,age\nAlice,30')
+  })
+
+  it('throws when the table belongs to a different workspace', async () => {
+    mockGetTableById.mockResolvedValue({ ...buildTable(), workspaceId: 'workspace-other' })
+
+    await expect(
+      executeFunctionExecute(
+        { code: 'print(1)', inputTables: ['tbl_1'] },
+        { userId: 'user-1', workflowId: 'wf-1', workspaceId: 'workspace-1' }
+      )
+    ).rejects.toThrow(/Input table not found/)
+    expect(mockExecuteTool).not.toHaveBeenCalled()
+  })
+})

apps/sim/lib/copilot/tools/handlers/function-execute.ts

@@ -3,7 +3,10 @@ import { decodeVfsPathSegments, encodeVfsPathSegments } from '@/lib/copilot/vfs/
 import { resolveWorkflowAliasForWorkspace } from '@/lib/copilot/vfs/workflow-alias-resolver'
 import { isPlanAliasPath, workflowAliasSandboxPath } from '@/lib/copilot/vfs/workflow-aliases'
 import { isMothershipBetaFeaturesEnabled } from '@/lib/core/config/feature-flags'
-import { getTableById, listTables, queryRows } from '@/lib/table/service'
+import { buildNameById, rowDataIdToName } from '@/lib/table/column-keys'
+import { toCsvRow } from '@/lib/table/export-format'
+import { getTableById, listTables, selectExportRowPage } from '@/lib/table/service'
+import type { TableDefinition } from '@/lib/table/types'
 import { listWorkspaceFileFolders } from '@/lib/uploads/contexts/workspace/workspace-file-folder-manager'
 import {
   fetchWorkspaceFileBuffer,
@@ -62,6 +65,44 @@ async function resolveTableRef(
   return tablePathLookup?.get(tableName) ?? null
 }
 
+const TABLE_MOUNT_PAGE_SIZE = 5000
+
+/**
+ * Serializes a cell for a sandbox CSV mount. Unlike export downloads this skips
+ * formula neutralization — the CSV is consumed by code, and a prefixed `'`
+ * would corrupt values.
+ */
+function formatMountCsvValue(value: unknown): string {
+  if (value === null || value === undefined) return ''
+  if (value instanceof Date) return value.toISOString()
+  if (typeof value === 'object') return JSON.stringify(value)
+  return String(value)
+}
+
+/**
+ * Serializes a full table to CSV for a sandbox mount. Walks the keyset export
+ * reader page by page so every row is included (`queryRows` with defaults
+ * silently truncated mounts to its 100-row page and paid for a count and
+ * execution metadata the CSV never used), and remaps stored column-id keys
+ * back to display names so headers line up with cell values.
+ */
+async function buildTableCsvForMount(table: TableDefinition): Promise<string> {
+  const nameById = buildNameById(table.schema)
+  const headers = table.schema.columns.map((c) => c.name)
+  const lines = [toCsvRow(headers)]
+  let after: { position: number; id: string } | null = null
+  while (true) {
+    const page = await selectExportRowPage(table, after, TABLE_MOUNT_PAGE_SIZE)
+    for (const row of page) {
+      const data = rowDataIdToName(row.data, nameById)
+      lines.push(toCsvRow(headers.map((header) => formatMountCsvValue(data[header]))))
+    }
+    if (page.length < TABLE_MOUNT_PAGE_SIZE) return lines.join('\n')
+    const last = page[page.length - 1]
+    after = { position: last.position, id: last.id }
+  }
+}
+
 async function resolveInputFiles(
   workspaceId: string,
   inputFiles?: unknown[],
@@ -247,55 +288,41 @@ async function resolveInputFiles(
     const tablePathLookup = hasTablePathRefs
       ? new Map((await listTables(workspaceId)).map((table) => [table.name, table]))
       : undefined
-    for (const tableRef of inputTables) {
-      const tableId =
-        typeof tableRef === 'string'
-          ? tableRef
-          : tableRef && typeof tableRef === 'object'
-            ? (tableRef as CanonicalTableInput).tableId || (tableRef as CanonicalTableInput).path
+    const tableMounts = await Promise.all(
+      inputTables.map(async (tableRef) => {
+        const tableId =
+          typeof tableRef === 'string'
+            ? tableRef
+            : tableRef && typeof tableRef === 'object'
+              ? (tableRef as CanonicalTableInput).tableId || (tableRef as CanonicalTableInput).path
+              : undefined
+        if (!tableId) return null
+        const table = await resolveTableRef(tableId, tablePathLookup)
+        if (!table || table.workspaceId !== workspaceId) {
+          throw new Error(
+            `Input table not found: "${tableId}". Pass the table id (tbl_...) from tables/{name}/meta.json, or a tables/{name}/meta.json path.`
+          )
+        }
+        const csvContent = await buildTableCsvForMount(table)
+        const sandboxPath =
+          typeof tableRef === 'object' && tableRef !== null
+            ? (tableRef as CanonicalTableInput).sandboxPath
             : undefined
-      if (!tableId) continue
-      const table = await resolveTableRef(tableId, tablePathLookup)
-      if (!table || table.workspaceId !== workspaceId) {
-        throw new Error(
-          `Input table not found: "${tableId}". Pass the table id (tbl_...) from tables/{name}/meta.json, or a tables/{name}/meta.json path.`
-        )
-      }
-      const rows = await queryRows(table, {}, 'copilot-fn-exec')
-
-      const allKeys = new Set(table.schema.columns.map((column) => column.name))
-      for (const row of rows.rows ?? []) {
-        if (row.data && typeof row.data === 'object') {
-          for (const key of Object.keys(row.data as Record<string, unknown>)) {
-            allKeys.add(key)
-          }
+        return {
+          path: sandboxPath || `/home/user/tables/${table.id}.csv`,
+          content: csvContent,
         }
-      }
-      const headers = Array.from(allKeys)
-      const csvLines = [headers.join(',')]
-      for (const row of rows.rows ?? []) {
-        const data = (row.data || {}) as Record<string, unknown>
-        csvLines.push(
-          headers
-            .map((h) => {
-              const val = data[h]
-              const str = val === null || val === undefined ? '' : String(val)
-              return str.includes(',') || str.includes('"') || str.includes('\n')
-                ? `"${str.replace(/"/g, '""')}"`
-                : str
-            })
-            .join(',')
+      })
+    )
+    for (const mount of tableMounts) {
+      if (!mount) continue
+      if (totalSize + mount.content.length > MAX_TOTAL_SIZE) {
+        throw new Error(
+          `Mounting table "${mount.path}" would exceed the ${MAX_TOTAL_SIZE / 1024 / 1024}MB total mount limit. Mount fewer or smaller tables.`
         )
       }
-      const csvContent = csvLines.join('\n')
-      const sandboxPath =
-        typeof tableRef === 'object' && tableRef !== null
-          ? (tableRef as CanonicalTableInput).sandboxPath
-          : undefined
-      sandboxFiles.push({
-        path: sandboxPath || `/home/user/tables/${table.id}.csv`,
-        content: csvContent,
-      })
+      totalSize += mount.content.length
+      sandboxFiles.push(mount)
     }
   }