feat(table): add workflow execution column type (#4338)

* Add table triggers for columns and row added

* Add async batching job for running column

* Add ui improvements, stop mechanism

* Use trigger dev for workflow runs

* Use unified column sidebar for table

* Add socket waits for tables, multi column workflow support

* change back to cell based trigger jobs

* Reuse code, add view log inline in table view

* reorganize db to treat each column separately

* adjust column naming strategy

* Column ui improvements

* fix live update on table

* Change new column behavior

* fixed errored workfows not showing as stopped

* Column sidebar improvements

* fix table column swapping behavior

* fix bugs

* add prompting, fix lint

* fix ui stuff

* flip feature flag

* Add zod contracts fix initial auto-run of columns

* ui improvements

* Use db filter to query unran rows

* Use live workflow run

* Change wording for deleting workflow column

* Update tools

* Add tool to run selected rows

* Add mothership tools

* adjust col width

* Restore ff

* fix drizzle migration

* fix test

Author: TheodoreSpeaks

apps/realtime/src/handlers/index.ts

@@ -2,6 +2,7 @@ import { setupConnectionHandlers } from '@/handlers/connection'
 import { setupOperationsHandlers } from '@/handlers/operations'
 import { setupPresenceHandlers } from '@/handlers/presence'
 import { setupSubblocksHandlers } from '@/handlers/subblocks'
+import { setupTableHandlers } from '@/handlers/tables'
 import { setupVariablesHandlers } from '@/handlers/variables'
 import { setupWorkflowHandlers } from '@/handlers/workflow'
 import type { AuthenticatedSocket } from '@/middleware/auth'
@@ -13,5 +14,6 @@ export function setupAllHandlers(socket: AuthenticatedSocket, roomManager: IRoom
   setupSubblocksHandlers(socket, roomManager)
   setupVariablesHandlers(socket, roomManager)
   setupPresenceHandlers(socket, roomManager)
+  setupTableHandlers(socket, roomManager)
   setupConnectionHandlers(socket, roomManager)
 }

apps/realtime/src/handlers/tables.ts

@@ -0,0 +1,73 @@
+import { createLogger } from '@sim/logger'
+import type { AuthenticatedSocket } from '@/middleware/auth'
+import { verifyTableAccess } from '@/middleware/permissions'
+import { type IRoomManager, tableRoomName } from '@/rooms/types'
+
+const logger = createLogger('TableHandlers')
+
+/**
+ * Wires `join-table` / `leave-table` socket events. Tables don't track presence
+ * or last-modified state — joining is a thin wrapper around `socket.join` so the
+ * Sim API → Realtime HTTP bridge can broadcast row updates back to subscribed clients.
+ */
+export function setupTableHandlers(socket: AuthenticatedSocket, _roomManager: IRoomManager) {
+  socket.on('join-table', async ({ tableId }: { tableId?: string }) => {
+    try {
+      if (!tableId || typeof tableId !== 'string') {
+        socket.emit('join-table-error', {
+          tableId: tableId ?? null,
+          error: 'tableId required',
+          code: 'INVALID_TABLE_ID',
+          retryable: false,
+        })
+        return
+      }
+
+      const userId = socket.userId
+      if (!userId) {
+        socket.emit('join-table-error', {
+          tableId,
+          error: 'Authentication required',
+          code: 'AUTHENTICATION_REQUIRED',
+          retryable: false,
+        })
+        return
+      }
+
+      const { hasAccess } = await verifyTableAccess(userId, tableId)
+      if (!hasAccess) {
+        socket.emit('join-table-error', {
+          tableId,
+          error: 'Access denied to table',
+          code: 'ACCESS_DENIED',
+          retryable: false,
+        })
+        return
+      }
+
+      const room = tableRoomName(tableId)
+      socket.join(room)
+      socket.emit('join-table-success', { tableId, socketId: socket.id })
+      logger.debug(`Socket ${socket.id} (user ${userId}) joined ${room}`)
+    } catch (error) {
+      logger.error(`Error joining table room:`, error)
+      socket.emit('join-table-error', {
+        tableId: null,
+        error: 'Failed to join table',
+        code: 'JOIN_TABLE_FAILED',
+        retryable: true,
+      })
+    }
+  })
+
+  socket.on('leave-table', async ({ tableId }: { tableId?: string }) => {
+    try {
+      if (!tableId || typeof tableId !== 'string') return
+      const room = tableRoomName(tableId)
+      socket.leave(room)
+      logger.debug(`Socket ${socket.id} left ${room}`)
+    } catch (error) {
+      logger.error(`Error leaving table room:`, error)
+    }
+  })
+}

apps/realtime/src/middleware/permissions.ts

@@ -131,3 +131,51 @@ export async function verifyWorkflowAccess(
     return { hasAccess: false }
   }
 }
+
+/**
+ * Verify a user has read access to a table by virtue of workspace permission.
+ * Mirrors `verifyWorkflowAccess` for the table-room socket join check.
+ */
+export async function verifyTableAccess(
+  userId: string,
+  tableId: string
+): Promise<{ hasAccess: boolean; workspaceId?: string }> {
+  try {
+    const { userTableDefinitions, permissions } = await import('@sim/db')
+    const tableData = await db
+      .select({ workspaceId: userTableDefinitions.workspaceId })
+      .from(userTableDefinitions)
+      .where(and(eq(userTableDefinitions.id, tableId), isNull(userTableDefinitions.archivedAt)))
+      .limit(1)
+
+    if (!tableData.length) {
+      logger.warn(`Table ${tableId} not found`)
+      return { hasAccess: false }
+    }
+    const { workspaceId } = tableData[0]
+    if (!workspaceId) return { hasAccess: false }
+
+    const [permissionRow] = await db
+      .select({ permissionType: permissions.permissionType })
+      .from(permissions)
+      .where(
+        and(
+          eq(permissions.userId, userId),
+          eq(permissions.entityType, 'workspace'),
+          eq(permissions.entityId, workspaceId)
+        )
+      )
+      .limit(1)
+
+    if (!permissionRow?.permissionType) {
+      logger.warn(
+        `User ${userId} has no permission for workspace ${workspaceId} (table ${tableId})`
+      )
+      return { hasAccess: false }
+    }
+    return { hasAccess: true, workspaceId }
+  } catch (error) {
+    logger.error(`Error verifying table access for user ${userId}, table ${tableId}:`, error)
+    return { hasAccess: false }
+  }
+}

apps/realtime/src/rooms/memory-manager.ts

@@ -1,6 +1,13 @@
 import { createLogger } from '@sim/logger'
 import type { Server } from 'socket.io'
-import type { IRoomManager, UserPresence, UserSession, WorkflowRoom } from '@/rooms/types'
+import {
+  type IRoomManager,
+  type TableRowUpdatedPayload,
+  tableRoomName,
+  type UserPresence,
+  type UserSession,
+  type WorkflowRoom,
+} from '@/rooms/types'
 
 const logger = createLogger('MemoryRoomManager')
 
@@ -255,4 +262,23 @@ export class MemoryRoomManager implements IRoomManager {
 
     logger.info(`Notified ${room.users.size} users about workflow deployment change: ${workflowId}`)
   }
+
+  emitToTable<T = unknown>(tableId: string, event: string, payload: T): void {
+    this._io.to(tableRoomName(tableId)).emit(event, payload)
+  }
+
+  async handleTableRowUpdated(tableId: string, payload: TableRowUpdatedPayload): Promise<void> {
+    this.emitToTable(tableId, 'table-row-updated', { tableId, ...payload })
+  }
+
+  async handleTableRowDeleted(tableId: string, rowId: string): Promise<void> {
+    this.emitToTable(tableId, 'table-row-deleted', { tableId, rowId })
+  }
+
+  async handleTableDeleted(tableId: string): Promise<void> {
+    logger.info(`Handling table deletion notification for ${tableId}`)
+    this.emitToTable(tableId, 'table-deleted', { tableId, timestamp: Date.now() })
+    // Eject sockets so they don't hold a stale room. Cross-pod safe via socket.io.
+    await this._io.in(tableRoomName(tableId)).socketsLeave(tableRoomName(tableId))
+  }
 }

apps/realtime/src/rooms/redis-manager.ts

@@ -1,7 +1,13 @@
 import { createLogger } from '@sim/logger'
 import { createClient, type RedisClientType } from 'redis'
 import type { Server } from 'socket.io'
-import type { IRoomManager, UserPresence, UserSession } from '@/rooms/types'
+import {
+  type IRoomManager,
+  type TableRowUpdatedPayload,
+  tableRoomName,
+  type UserPresence,
+  type UserSession,
+} from '@/rooms/types'
 
 const logger = createLogger('RedisRoomManager')
 
@@ -457,4 +463,23 @@ export class RedisRoomManager implements IRoomManager {
     const userCount = await this.getUniqueUserCount(workflowId)
     logger.info(`Notified ${userCount} users about workflow deployment change: ${workflowId}`)
   }
+
+  emitToTable<T = unknown>(tableId: string, event: string, payload: T): void {
+    this._io.to(tableRoomName(tableId)).emit(event, payload)
+  }
+
+  async handleTableRowUpdated(tableId: string, payload: TableRowUpdatedPayload): Promise<void> {
+    this.emitToTable(tableId, 'table-row-updated', { tableId, ...payload })
+  }
+
+  async handleTableRowDeleted(tableId: string, rowId: string): Promise<void> {
+    this.emitToTable(tableId, 'table-row-deleted', { tableId, rowId })
+  }
+
+  async handleTableDeleted(tableId: string): Promise<void> {
+    logger.info(`Handling table deletion notification for ${tableId}`)
+    this.emitToTable(tableId, 'table-deleted', { tableId, timestamp: Date.now() })
+    // Eject sockets across all pods via socket.io's Redis adapter.
+    await this._io.in(tableRoomName(tableId)).socketsLeave(tableRoomName(tableId))
+  }
 }

apps/realtime/src/rooms/types.ts

@@ -143,4 +143,45 @@ export interface IRoomManager {
    * Handle workflow deployment change - notify users to refresh deployment state
    */
   handleWorkflowDeployed(workflowId: string): Promise<void>
+
+  /**
+   * Emit an event to all clients in a table room (`table:${tableId}`).
+   * Tables don't track presence/last-modified state — just pub/sub.
+   */
+  emitToTable<T = unknown>(tableId: string, event: string, payload: T): void
+
+  /**
+   * Notify all clients in a table room of a row write (insert/update/cell-state-change).
+   * Sim API calls this via the `/api/table-row-updated` HTTP bridge after every successful
+   * row commit; the client merges the delta into its React Query cache.
+   */
+  handleTableRowUpdated(tableId: string, payload: TableRowUpdatedPayload): Promise<void>
+
+  /**
+   * Notify all clients in a table room that a row has been deleted.
+   */
+  handleTableRowDeleted(tableId: string, rowId: string): Promise<void>
+
+  /**
+   * Notify all clients in a table room that the table has been deleted; eject sockets.
+   */
+  handleTableDeleted(tableId: string): Promise<void>
+}
+
+/**
+ * Payload broadcast on `table-row-updated`. Mirrors the shape of `TableRow.data` so
+ * the client can merge directly into its React Query rows cache. `position` and
+ * `updatedAt` are included for cache reconciliation; `data` is the full row data
+ * (not a per-cell delta) — see plan Notes.
+ */
+export interface TableRowUpdatedPayload {
+  rowId: string
+  data: Record<string, unknown>
+  /** Per-workflow-group execution state. Keyed by `WorkflowGroup.id`. */
+  executions?: Record<string, unknown>
+  position: number
+  updatedAt: string | number
 }
+
+/** Socket.IO room name for a table. Namespaced from workflow rooms. */
+export const tableRoomName = (tableId: string): string => `table:${tableId}`

apps/realtime/src/routes/http.ts

@@ -150,6 +150,52 @@ export function createHttpHandler(roomManager: IRoomManager, logger: Logger) {
       return
     }
 
+    // Handle table row write notifications from the Sim API
+    if (req.method === 'POST' && req.url === '/api/table-row-updated') {
+      try {
+        const body = await readRequestBody(req)
+        const { tableId, rowId, data, executions, position, updatedAt } = JSON.parse(body)
+        await roomManager.handleTableRowUpdated(tableId, {
+          rowId,
+          data,
+          executions,
+          position,
+          updatedAt,
+        })
+        sendSuccess(res)
+      } catch (error) {
+        logger.error('Error handling table row update notification:', error)
+        sendError(res, 'Failed to process table row update')
+      }
+      return
+    }
+
+    if (req.method === 'POST' && req.url === '/api/table-row-deleted') {
+      try {
+        const body = await readRequestBody(req)
+        const { tableId, rowId } = JSON.parse(body)
+        await roomManager.handleTableRowDeleted(tableId, rowId)
+        sendSuccess(res)
+      } catch (error) {
+        logger.error('Error handling table row deletion notification:', error)
+        sendError(res, 'Failed to process table row deletion')
+      }
+      return
+    }
+
+    if (req.method === 'POST' && req.url === '/api/table-deleted') {
+      try {
+        const body = await readRequestBody(req)
+        const { tableId } = JSON.parse(body)
+        await roomManager.handleTableDeleted(tableId)
+        sendSuccess(res)
+      } catch (error) {
+        logger.error('Error handling table deletion notification:', error)
+        sendError(res, 'Failed to process table deletion')
+      }
+      return
+    }
+
     res.writeHead(404, { 'Content-Type': 'application/json' })
     res.end(JSON.stringify({ error: 'Not found' }))
   }

apps/sim/app/api/table/[tableId]/cancel-runs/route.ts

@@ -0,0 +1,57 @@
+import { createLogger } from '@sim/logger'
+import { type NextRequest, NextResponse } from 'next/server'
+import { cancelTableRunsContract } from '@/lib/api/contracts/tables'
+import { parseRequest } from '@/lib/api/server'
+import { checkSessionOrInternalAuth } from '@/lib/auth/hybrid'
+import { generateRequestId } from '@/lib/core/utils/request'
+import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
+import { cancelWorkflowGroupRuns } from '@/lib/table/workflow-columns'
+import { accessError, checkAccess } from '@/app/api/table/utils'
+
+const logger = createLogger('TableCancelRunsAPI')
+
+interface RouteParams {
+  params: Promise<{ tableId: string }>
+}
+
+/**
+ * POST /api/table/[tableId]/cancel-runs
+ *
+ * Cancels in-flight and pending workflow-column runs for this table. Scopes:
+ * `all` (every cell) or `row` (every cell for `rowId`).
+ */
+export const POST = withRouteHandler(async (request: NextRequest, { params }: RouteParams) => {
+  const requestId = generateRequestId()
+
+  try {
+    const authResult = await checkSessionOrInternalAuth(request, { requireWorkflowId: false })
+    if (!authResult.success || !authResult.userId) {
+      return NextResponse.json({ error: 'Authentication required' }, { status: 401 })
+    }
+
+    const parsed = await parseRequest(cancelTableRunsContract, request, { params })
+    if (!parsed.success) return parsed.response
+    const { tableId } = parsed.data.params
+    const { workspaceId, scope, rowId } = parsed.data.body
+
+    const result = await checkAccess(tableId, authResult.userId, 'write')
+    if (!result.ok) return accessError(result, requestId, tableId)
+    const { table } = result
+
+    if (table.workspaceId !== workspaceId) {
+      return NextResponse.json({ error: 'Invalid workspace ID' }, { status: 400 })
+    }
+
+    const cancelled = await cancelWorkflowGroupRuns(tableId, scope === 'row' ? rowId : undefined)
+    logger.info(
+      `[${requestId}] cancel-runs: tableId=${tableId} scope=${scope}${
+        rowId ? ` rowId=${rowId}` : ''
+      } cancelled=${cancelled}`
+    )
+
+    return NextResponse.json({ success: true, data: { cancelled } })
+  } catch (error) {
+    logger.error(`[${requestId}] cancel-runs failed:`, error)
+    return NextResponse.json({ error: 'Failed to cancel runs' }, { status: 500 })
+  }
+})