diff --git a/apps/sim/background/cleanup-table-row-ttl.test.ts b/apps/sim/background/cleanup-table-row-ttl.test.ts
index 9a4db2c92cc..d49cf34eae5 100644
--- a/apps/sim/background/cleanup-table-row-ttl.test.ts
+++ b/apps/sim/background/cleanup-table-row-ttl.test.ts
@@ -15,6 +15,7 @@ const {
   mockSignalTableRowsChanged,
   mockTask,
   mockWithLockedTable,
+  mockFireTableTrigger,
 } = vi.hoisted(() => ({
   mockDeleteExecute: vi.fn(),
   mockListExecute: vi.fn(),
@@ -22,6 +23,7 @@ const {
   mockSignalTableRowsChanged: vi.fn(),
   mockTask: vi.fn((config: unknown) => config),
   mockWithLockedTable: vi.fn(),
+  mockFireTableTrigger: vi.fn(),
 }))
 
 vi.mock('@sim/db', () => ({
@@ -30,10 +32,12 @@ vi.mock('@sim/db', () => ({
 
 vi.mock('@trigger.dev/sdk', () => ({ task: mockTask }))
 vi.mock('@/lib/table/events', () => ({ signalTableRowsChanged: mockSignalTableRowsChanged }))
+vi.mock('@/lib/table/constants', () => ({ getDeleteSnapshotBatchSize: () => 500 }))
 vi.mock('@/lib/table/service', () => ({ withLockedTable: mockWithLockedTable }))
 vi.mock('@/lib/table/ttl-availability', () => ({
   isTableRowTtlEnabled: mockIsTableRowTtlEnabled,
 }))
+vi.mock('@/lib/table/trigger', () => ({ fireTableTrigger: mockFireTableTrigger }))
 
 import { cleanupTableRowTtlTask, runCleanupTableRowTtl } from '@/background/cleanup-table-row-ttl'
 
@@ -41,11 +45,26 @@ const dialect = new PgDialect()
 
 const table = {
   id: 'table-1',
+  name: 'Expiring rows',
   workspaceId: 'workspace-1',
   schema: { columns: [{ id: 'col-ttl', name: 'expires_at', type: 'ttl' }] },
   locks: { insertLocked: false, updateLocked: false, deleteLocked: false, schemaLocked: false },
 }
 
+function deletedRows(count: number, start = 1) {
+  return Array.from({ length: count }, (_, index) => {
+    const number = start + index
+    return { id: `row-${number}`, data: { value: number } }
+  })
+}
+
+function returnedRows(count: number, start = 1, createdAt = '2026-01-01T00:00:00.000000') {
+  return deletedRows(count, start).map((row) => ({
+    ...row,
+    createdAt,
+  }))
+}
+
 describe('table row TTL cleanup', () => {
   beforeEach(() => {
     vi.clearAllMocks()
@@ -65,11 +84,10 @@ describe('table row TTL cleanup', () => {
   it('deletes expired rows in locked, created-at keyset batches and signals the table', async () => {
     mockDeleteExecute
       .mockResolvedValueOnce([
-        { count: 500, createdAt: '2026-01-01T00:00:00.123456', lastId: 'row-500' },
-      ])
-      .mockResolvedValueOnce([
-        { count: 12, createdAt: '2026-01-02T00:00:00.000000', lastId: 'row-512' },
+        ...returnedRows(499, 1, '2026-01-01T00:00:00.123455'),
+        ...returnedRows(1, 500, '2026-01-01T00:00:00.123456'),
       ])
+      .mockResolvedValueOnce(returnedRows(12, 501))
 
     await expect(runCleanupTableRowTtl()).resolves.toEqual({
       batches: 2,
@@ -86,13 +104,25 @@ describe('table row TTL cleanup', () => {
       expect.arrayContaining(['2026-01-01T00:00:00.123456', 'row-500'])
     )
     expect(mockSignalTableRowsChanged).toHaveBeenCalledWith(table.id)
+    expect(mockFireTableTrigger).toHaveBeenCalledTimes(2)
+    expect(mockFireTableTrigger).toHaveBeenNthCalledWith(
+      1,
+      table.id,
+      table.workspaceId,
+      table.name,
+      'delete',
+      deletedRows(500),
+      null,
+      table.schema,
+      'ttl-cleanup'
+    )
   })
 
   it('compares TTL values with whole Date.now epoch seconds', async () => {
     const nowEpochMilliseconds = 1_700_000_000_999
     const nowEpochSeconds = 1_700_000_000
     const nowSpy = vi.spyOn(Date, 'now').mockReturnValue(nowEpochMilliseconds)
-    mockDeleteExecute.mockResolvedValue([{ count: 0, createdAt: null, lastId: null }])
+    mockDeleteExecute.mockResolvedValue([])
 
     try {
       await runCleanupTableRowTtl()
@@ -109,7 +139,7 @@ describe('table row TTL cleanup', () => {
   })
 
   it('checks the oldest expired rows first without using creation time as an expiry rule', async () => {
-    mockDeleteExecute.mockResolvedValue([{ count: 0, createdAt: null, lastId: null }])
+    mockDeleteExecute.mockResolvedValue([])
 
     await runCleanupTableRowTtl()
 
@@ -120,12 +150,14 @@ describe('table row TTL cleanup', () => {
       .trim()
     expect(query).toContain('AND (table_row.data->>?)::numeric <= ?')
     expect(query).toContain('ORDER BY table_row.created_at, table_row.id')
-    expect(query).toContain(`to_char(created_at, 'YYYY-MM-DD"T"HH24:MI:SS.US')`)
+    expect(query).toContain(
+      `to_char(table_row.created_at, 'YYYY-MM-DD"T"HH24:MI:SS.US') AS "createdAt"`
+    )
     expect(query).not.toContain('table_row.created_by')
   })
 
   it('rejects a batch without a creation-time cursor', async () => {
-    mockDeleteExecute.mockResolvedValue([{ count: 1, lastId: 'row-1' }])
+    mockDeleteExecute.mockResolvedValue([{ id: 'row-1', data: { value: 1 } }])
 
     await expect(runCleanupTableRowTtl()).rejects.toThrow(
       'Table row TTL cleanup did not return a creation-time cursor'
@@ -174,9 +206,7 @@ describe('table row TTL cleanup', () => {
   })
 
   it('stops after one hundred full batches', async () => {
-    mockDeleteExecute.mockResolvedValue([
-      { count: 500, createdAt: '2026-01-01T00:00:00.000000', lastId: 'row-cursor' },
-    ])
+    mockDeleteExecute.mockResolvedValue(returnedRows(500))
 
     await expect(runCleanupTableRowTtl()).resolves.toEqual({
       batches: 100,
@@ -206,12 +236,12 @@ describe('table row TTL cleanup', () => {
           const attempt = (tableAttempts.get(tableId) ?? 0) + 1
           tableAttempts.set(tableId, attempt)
           if (tableId === table.id && attempt === 1) {
-            return [{ count: 500, createdAt: '2026-01-01T00:00:00.000000', lastId: 'row-500' }]
+            return returnedRows(500)
           }
           if (tableId === secondTable.id) {
-            return [{ count: 1, createdAt: '2026-01-01T00:00:00.000000', lastId: 'row-1' }]
+            return returnedRows(1)
           }
-          return [{ count: 0, createdAt: null, lastId: null }]
+          return []
         }),
       })
     })
diff --git a/apps/sim/background/cleanup-table-row-ttl.ts b/apps/sim/background/cleanup-table-row-ttl.ts
index 65784cc3324..bd343a71970 100644
--- a/apps/sim/background/cleanup-table-row-ttl.ts
+++ b/apps/sim/background/cleanup-table-row-ttl.ts
@@ -5,16 +5,19 @@ import { task } from '@trigger.dev/sdk'
 import { sql } from 'drizzle-orm'
 import { asOrchestrationError } from '@/lib/core/orchestration/types'
 import { getColumnId } from '@/lib/table/column-keys'
+import { getDeleteSnapshotBatchSize } from '@/lib/table/constants'
 import { signalTableRowsChanged } from '@/lib/table/events'
 import { assertRowDelete, TableLockedError } from '@/lib/table/mutation-locks'
 import type { DbTransaction } from '@/lib/table/planner'
+import type { DeletedTableRow } from '@/lib/table/rows/ordering'
 import { withLockedTable } from '@/lib/table/service'
+import { fireTableTrigger } from '@/lib/table/trigger'
 import { isTableRowTtlEnabled } from '@/lib/table/ttl-availability'
+import type { RowData, TableSchema } from '@/lib/table/types'
 
 const logger = createLogger('CleanupTableRowTtl')
 const cleanupDb = dbFor('cleanup')
 
-const TTL_CLEANUP_BATCH_SIZE = 500
 const TTL_CLEANUP_MAX_BATCHES = 100
 
 interface ExpiredTtlTableRef {
@@ -23,12 +26,20 @@ interface ExpiredTtlTableRef {
   workspaceId: string
 }
 
-interface DeletedTtlBatch {
-  attempted: boolean
+interface DeletedTtlRows {
   deleted: number
   cursor: TtlCleanupCursor | null
+  rows: DeletedTableRow[]
 }
 
+type DeletedTtlBatch =
+  | { attempted: false; deleted: 0; cursor: null }
+  | (DeletedTtlRows & {
+      attempted: true
+      tableName: string
+      schema: TableSchema
+    })
+
 interface TtlCleanupCursor {
   createdAt: string
   id: string
@@ -85,34 +96,30 @@ async function listExpiredTtlTables(nowEpochSeconds: number): Promise<ExpiredTtl
   return Array.isArray(rows) ? rows : []
 }
 
-function parseDeletedBatch(rows: unknown): Omit<DeletedTtlBatch, 'attempted'> {
-  const [row] = Array.isArray(rows)
-    ? (rows as Array<{
-        count?: number | string
-        createdAt?: string | null
-        lastId?: string | null
-      }>)
-    : []
-  if (!row) throw new Error('Table row TTL cleanup did not return a deleted count')
-
-  const deleted = Number(row.count)
-  if (!Number.isSafeInteger(deleted) || deleted < 0 || deleted > TTL_CLEANUP_BATCH_SIZE) {
+function parseDeletedBatch(rows: unknown, batchSize: number): DeletedTtlRows {
+  if (!Array.isArray(rows)) {
+    throw new Error('Table row TTL cleanup did not return deleted rows')
+  }
+  const deletedRows = rows as Array<{ id?: unknown; data?: unknown; createdAt?: unknown }>
+  if (deletedRows.length > batchSize) {
     throw new Error('Table row TTL cleanup returned an invalid deleted count')
   }
-  if (deleted > 0) {
-    if (typeof row.lastId !== 'string') {
+  const parsed = deletedRows.map((row) => {
+    if (typeof row.id !== 'string') {
       throw new Error('Table row TTL cleanup did not return a row cursor')
     }
     if (typeof row.createdAt !== 'string') {
       throw new Error('Table row TTL cleanup did not return a creation-time cursor')
     }
-  }
+    return {
+      cursor: { createdAt: row.createdAt, id: row.id },
+      row: { id: row.id, data: row.data as RowData },
+    }
+  })
   return {
-    deleted,
-    cursor:
-      typeof row.createdAt === 'string' && typeof row.lastId === 'string'
-        ? { createdAt: row.createdAt, id: row.lastId }
-        : null,
+    deleted: parsed.length,
+    cursor: parsed[parsed.length - 1]?.cursor ?? null,
+    rows: parsed.map(({ row }) => row),
   }
 }
 
@@ -122,13 +129,10 @@ async function deleteExpiredTableRowBatch(
   workspaceId: string,
   columnKey: string,
   nowEpochSeconds: number,
+  batchSize: number,
   after?: TtlCleanupCursor
-): Promise<Omit<DeletedTtlBatch, 'attempted'>> {
-  const rows = await trx.execute<{
-    count: number | string
-    createdAt: string | null
-    lastId: string | null
-  }>(sql`
+): Promise<DeletedTtlRows> {
+  const rows = await trx.execute<{ id: string; data: RowData; createdAt: string }>(sql`
     WITH candidates AS MATERIALIZED (
       SELECT table_row.id
       FROM ${userTableRows} AS table_row
@@ -142,37 +146,34 @@ async function deleteExpiredTableRowBatch(
         AND jsonb_typeof(table_row.data->${columnKey}) = 'number'
         AND (table_row.data->>${columnKey})::numeric <= ${nowEpochSeconds}
       ORDER BY table_row.created_at, table_row.id
-      LIMIT ${TTL_CLEANUP_BATCH_SIZE}
+      LIMIT ${batchSize}
       FOR UPDATE OF table_row SKIP LOCKED
     ), deleted AS (
       DELETE FROM ${userTableRows} AS table_row
       USING candidates
       WHERE table_row.id = candidates.id
-      RETURNING table_row.id, table_row.created_at
+      RETURNING
+        table_row.id,
+        table_row.data,
+        to_char(table_row.created_at, 'YYYY-MM-DD"T"HH24:MI:SS.US') AS "createdAt"
     )
-    SELECT
-      count(*)::integer AS count,
-      (array_agg(id ORDER BY created_at DESC, id DESC))[1] AS "lastId",
-      (
-        array_agg(
-          to_char(created_at, 'YYYY-MM-DD"T"HH24:MI:SS.US')
-          ORDER BY created_at DESC, id DESC
-        )
-      )[1] AS "createdAt"
+    SELECT id, data, "createdAt"
     FROM deleted
+    ORDER BY "createdAt", id
   `)
-  return parseDeletedBatch(rows)
+  return parseDeletedBatch(rows, batchSize)
 }
 
 async function deleteExpiredRowsForTable(
   ref: ExpiredTtlTableRef,
   nowEpochSeconds: number,
+  batchSize: number,
   after?: TtlCleanupCursor
 ): Promise<DeletedTtlBatch> {
   try {
-    return await withLockedTable(
+    const batch = await withLockedTable(
       ref.id,
-      async (table, trx) => {
+      async (table, trx): Promise<DeletedTtlBatch> => {
         try {
           assertRowDelete(table)
         } catch (error) {
@@ -191,12 +192,31 @@ async function deleteExpiredRowsForTable(
           table.workspaceId,
           getColumnId(ttlColumn),
           nowEpochSeconds,
+          batchSize,
           after
         )
-        return { attempted: true, ...batch }
+        return {
+          attempted: true,
+          ...batch,
+          tableName: table.name,
+          schema: table.schema,
+        } satisfies DeletedTtlBatch
       },
       { expectedWorkspaceId: ref.workspaceId }
     )
+    if (batch.attempted && batch.rows.length > 0) {
+      await fireTableTrigger(
+        ref.id,
+        ref.workspaceId,
+        batch.tableName,
+        'delete',
+        batch.rows,
+        null,
+        batch.schema,
+        'ttl-cleanup'
+      )
+    }
+    return batch
   } catch (error) {
     if (asOrchestrationError(error)?.code === 'not_found') {
       return { attempted: false, deleted: 0, cursor: null }
@@ -216,6 +236,7 @@ export async function runCleanupTableRowTtl(
   }
 
   const nowEpochSeconds = Math.floor(Date.now() / 1000)
+  const batchSize = getDeleteSnapshotBatchSize()
   const tableRefs = await listExpiredTtlTables(nowEpochSeconds)
   const tableStates: TtlTableCleanupState[] = tableRefs.map((ref) => ({
     ref,
@@ -234,7 +255,12 @@ export async function runCleanupTableRowTtl(
       if (state.complete) continue
       if (batches === TTL_CLEANUP_MAX_BATCHES || signal?.aborted) break
 
-      const batch = await deleteExpiredRowsForTable(state.ref, nowEpochSeconds, state.after)
+      const batch = await deleteExpiredRowsForTable(
+        state.ref,
+        nowEpochSeconds,
+        batchSize,
+        state.after
+      )
       if (!batch.attempted) {
         state.complete = true
         continue
@@ -244,7 +270,7 @@ export async function runCleanupTableRowTtl(
       deleted += batch.deleted
       state.deleted += batch.deleted
       state.after = batch.cursor ?? undefined
-      if (batch.deleted < TTL_CLEANUP_BATCH_SIZE) state.complete = true
+      if (batch.deleted < batchSize) state.complete = true
     }
   }
 
diff --git a/apps/sim/lib/table/__tests__/service-filter-threading.test.ts b/apps/sim/lib/table/__tests__/service-filter-threading.test.ts
index 52ae732e805..e5b89dc7239 100644
--- a/apps/sim/lib/table/__tests__/service-filter-threading.test.ts
+++ b/apps/sim/lib/table/__tests__/service-filter-threading.test.ts
@@ -15,6 +15,8 @@ import { decodeCursor } from '@/lib/table/rows/cursor'
 import { buildFilterClause, buildSortClause } from '@/lib/table/sql'
 import type { ColumnDefinition, TableDefinition } from '@/lib/table/types'
 
+const { mockFireTableTrigger } = vi.hoisted(() => ({ mockFireTableTrigger: vi.fn() }))
+
 vi.mock('@/lib/table/sql', () => ({
   buildFilterClause: vi.fn(() => sql`true`),
   buildSortClause: vi.fn(() => sql`true`),
@@ -23,7 +25,7 @@ vi.mock('@/lib/table/sql', () => ({
 }))
 
 vi.mock('@/lib/table/trigger', () => ({
-  fireTableTrigger: vi.fn(),
+  fireTableTrigger: mockFireTableTrigger,
 }))
 
 vi.mock('@/lib/table/workflow-group-deps', () => ({
@@ -64,7 +66,9 @@ vi.mock('@/lib/table/validation', () => ({
 }))
 
 import {
+  deleteRow,
   deleteRowsByFilter,
+  deleteRowsByIds,
   queryRows,
   requireTableRowIds,
   updateRowsByFilter,
@@ -183,6 +187,82 @@ describe('service filter threading', () => {
   })
 })
 
+describe('delete trigger dispatch', () => {
+  beforeEach(() => {
+    vi.clearAllMocks()
+    resetDbChainMock()
+  })
+
+  it('fires with the committed snapshot after deleting one row', async () => {
+    dbChainMockFns.returning.mockResolvedValueOnce([{ id: 'row-1', data: { name: 'Ada' } }])
+
+    await deleteRow(TABLE, 'row-1', 'req-delete-one')
+
+    expect(mockFireTableTrigger).toHaveBeenCalledWith(
+      TABLE.id,
+      TABLE.workspaceId,
+      TABLE.name,
+      'delete',
+      [{ id: 'row-1', data: { name: 'Ada' } }],
+      null,
+      TABLE.schema,
+      'req-delete-one'
+    )
+  })
+
+  it('fires once with every committed snapshot in an ID batch', async () => {
+    dbChainMockFns.returning.mockResolvedValueOnce([
+      { id: 'row-1', data: { name: 'Ada' } },
+      { id: 'row-2', data: { name: 'Grace' } },
+    ])
+
+    await deleteRowsByIds(
+      TABLE,
+      { tableId: TABLE.id, workspaceId: TABLE.workspaceId, rowIds: ['row-1', 'row-2'] },
+      'req-delete-many'
+    )
+
+    expect(mockFireTableTrigger).toHaveBeenCalledWith(
+      TABLE.id,
+      TABLE.workspaceId,
+      TABLE.name,
+      'delete',
+      [
+        { id: 'row-1', data: { name: 'Ada' } },
+        { id: 'row-2', data: { name: 'Grace' } },
+      ],
+      null,
+      TABLE.schema,
+      'req-delete-many'
+    )
+  })
+
+  it('dispatches byte-bounded ID-delete snapshots before loading the next batch', async () => {
+    setEnv({
+      TABLE_MAX_ROW_SIZE_BYTES: TABLE_LIMITS.DELETE_SNAPSHOT_BATCH_MAX_BYTES * 2,
+    })
+    dbChainMockFns.returning
+      .mockResolvedValueOnce([{ id: 'row-1', data: { name: 'Ada' } }])
+      .mockResolvedValueOnce([{ id: 'row-2', data: { name: 'Grace' } }])
+
+    try {
+      await deleteRowsByIds(
+        TABLE,
+        { tableId: TABLE.id, workspaceId: TABLE.workspaceId, rowIds: ['row-1', 'row-2'] },
+        'req-delete-bounded'
+      )
+    } finally {
+      setEnv({ TABLE_MAX_ROW_SIZE_BYTES: undefined })
+    }
+
+    expect(mockFireTableTrigger).toHaveBeenCalledTimes(2)
+    expect(mockFireTableTrigger.mock.calls[0][4]).toEqual([{ id: 'row-1', data: { name: 'Ada' } }])
+    expect(mockFireTableTrigger.mock.calls[1][4]).toEqual([
+      { id: 'row-2', data: { name: 'Grace' } },
+    ])
+  })
+})
+
 describe('bulk update/delete limited-subset ordering', () => {
   beforeEach(() => {
     vi.clearAllMocks()
diff --git a/apps/sim/lib/table/__tests__/trigger.test.ts b/apps/sim/lib/table/__tests__/trigger.test.ts
index 64b5c4afc5e..fc04df054f1 100644
--- a/apps/sim/lib/table/__tests__/trigger.test.ts
+++ b/apps/sim/lib/table/__tests__/trigger.test.ts
@@ -60,7 +60,7 @@ interface Payload {
 function webhookEntry(config: Record<string, unknown> = {}) {
   return {
     webhook: { id: 'wh_1', providerConfig: { tableId: 'tbl_1', eventType: 'insert', ...config } },
-    workflow: { id: 'wf_1' },
+    workflow: { id: 'wf_1', workspaceId: 'ws_1' },
   }
 }
 
@@ -69,12 +69,13 @@ function firedPayloads(): Payload[] {
 }
 
 async function fire(
-  eventType: 'insert' | 'update',
+  eventType: 'insert' | 'update' | 'delete',
   data: RowData,
   oldRows: Map<string, RowData> | null = null
 ) {
   await fireTableTrigger(
     'tbl_1',
+    'ws_1',
     'Issues',
     eventType,
     [{ id: 'row_1', data } as never],
@@ -166,6 +167,19 @@ describe('fireTableTrigger — payload shape', () => {
     const [payload] = firedPayloads()
     expect(payload.changedColumns).toEqual(['Status'])
   })
+
+  it('emits the deleted row snapshot as the event row and previous row', async () => {
+    mockFetchActiveWebhooks.mockResolvedValue([webhookEntry({ eventType: 'delete' })])
+
+    await fire('delete', { col_title: 'Removed issue', col_status: 'opt_closed' })
+
+    const [payload] = firedPayloads()
+    const deletedRow = { Title: 'Removed issue', Status: 'Closed' }
+    expect(payload.rawRow).toEqual(deletedRow)
+    expect(payload.row).toEqual({ ...deletedRow, Tags: null })
+    expect(payload.previousRow).toEqual(deletedRow)
+    expect(payload.changedColumns).toEqual([])
+  })
 })
 
 describe('fireTableTrigger — gating', () => {
@@ -180,6 +194,14 @@ describe('fireTableTrigger — gating', () => {
     expect(mockProcessPolledWebhookEvent).not.toHaveBeenCalled()
   })
 
+  it('fires nothing for a workflow in a different workspace', async () => {
+    mockFetchActiveWebhooks.mockResolvedValue([
+      { ...webhookEntry(), workflow: { id: 'wf_other', workspaceId: 'ws_other' } },
+    ])
+    await fire('insert', { col_title: 'x' })
+    expect(mockProcessPolledWebhookEvent).not.toHaveBeenCalled()
+  })
+
   it('fires nothing when the event type does not match', async () => {
     mockFetchActiveWebhooks.mockResolvedValue([webhookEntry({ eventType: 'update' })])
     await fire('insert', { col_title: 'x' })
diff --git a/apps/sim/lib/table/constants.test.ts b/apps/sim/lib/table/constants.test.ts
index d3bd1ef4848..68d0b1c723b 100644
--- a/apps/sim/lib/table/constants.test.ts
+++ b/apps/sim/lib/table/constants.test.ts
@@ -39,6 +39,7 @@ declare module '@/lib/table/constants?constants-test' {
 
 import {
   getBillingDisabledTableLimits,
+  getDeleteSnapshotBatchSize,
   getMaxPageBytes,
   TABLE_LIMITS,
 } from '@/lib/table/constants?constants-test'
@@ -86,3 +87,23 @@ describe('getMaxPageBytes', () => {
     expect(getMaxPageBytes()).toBe(2 * 1024 * 1024)
   })
 })
+
+describe('getDeleteSnapshotBatchSize', () => {
+  beforeEach(() => {
+    for (const key of Object.keys(mockEnv)) delete mockEnv[key]
+  })
+
+  it('derives a worst-case row cap from the delete snapshot byte budget', () => {
+    expect(getDeleteSnapshotBatchSize()).toBe(
+      Math.floor(TABLE_LIMITS.DELETE_SNAPSHOT_BATCH_MAX_BYTES / TABLE_LIMITS.MAX_ROW_SIZE_BYTES)
+    )
+  })
+
+  it('always processes one row and never exceeds the delete row-count cap', () => {
+    mockEnv.TABLE_MAX_ROW_SIZE_BYTES = String(TABLE_LIMITS.DELETE_SNAPSHOT_BATCH_MAX_BYTES * 2)
+    expect(getDeleteSnapshotBatchSize()).toBe(1)
+
+    mockEnv.TABLE_MAX_ROW_SIZE_BYTES = '1'
+    expect(getDeleteSnapshotBatchSize()).toBe(TABLE_LIMITS.DELETE_BATCH_SIZE)
+  })
+})
diff --git a/apps/sim/lib/table/constants.ts b/apps/sim/lib/table/constants.ts
index 2da9bf6b58a..e5c514acd8e 100644
--- a/apps/sim/lib/table/constants.ts
+++ b/apps/sim/lib/table/constants.ts
@@ -39,6 +39,12 @@ export const TABLE_LIMITS = {
   UPDATE_BATCH_SIZE: 100,
   /** Batch size for bulk delete operations */
   DELETE_BATCH_SIZE: 1000,
+  /**
+   * Maximum serialized row-data bytes returned from one committed delete for
+   * post-commit trigger dispatch. The row-count cap is derived from this budget
+   * and the configured maximum row size before a DELETE materializes snapshots.
+   */
+  DELETE_SNAPSHOT_BATCH_MAX_BYTES: 32 * 1024 * 1024,
   /** Maximum rows per batch insert */
   MAX_BATCH_INSERT_SIZE: 1000,
   /** Maximum rows per bulk update/delete operation */
@@ -149,6 +155,21 @@ export function getMaxRowSizeBytes(): number {
   })
 }
 
+/**
+ * Maximum rows one delete may materialize with their JSON data for trigger
+ * dispatch. Uses the worst-case configured row size so every batch has an
+ * explicit byte bound before PostgreSQL returns it to the app process.
+ */
+export function getDeleteSnapshotBatchSize(): number {
+  return Math.max(
+    1,
+    Math.min(
+      TABLE_LIMITS.DELETE_BATCH_SIZE,
+      Math.floor(TABLE_LIMITS.DELETE_SNAPSHOT_BATCH_MAX_BYTES / getMaxRowSizeBytes())
+    )
+  )
+}
+
 export type PlanName = keyof typeof DEFAULT_TABLE_PLAN_LIMITS
 
 export interface TablePlanLimits {
diff --git a/apps/sim/lib/table/delete-runner.test.ts b/apps/sim/lib/table/delete-runner.test.ts
index aa19a0faac4..8eb163aeed0 100644
--- a/apps/sim/lib/table/delete-runner.test.ts
+++ b/apps/sim/lib/table/delete-runner.test.ts
@@ -16,6 +16,7 @@ const {
   mockAppendTableEvent,
   mockSignalTableRowsChanged,
   mockBuildFilterClause,
+  mockFireTableTrigger,
 } = vi.hoisted(() => ({
   mockGetTableById: vi.fn(),
   mockGetJobProgress: vi.fn(),
@@ -28,6 +29,7 @@ const {
   mockAppendTableEvent: vi.fn(),
   mockSignalTableRowsChanged: vi.fn(),
   mockBuildFilterClause: vi.fn(),
+  mockFireTableTrigger: vi.fn(),
 }))
 
 vi.mock('@/lib/table/service', () => ({
@@ -49,6 +51,7 @@ vi.mock('@/lib/table/events', () => ({
   signalTableRowsChanged: mockSignalTableRowsChanged,
 }))
 vi.mock('@/lib/table/sql', () => ({ buildFilterClause: mockBuildFilterClause }))
+vi.mock('@/lib/table/trigger', () => ({ fireTableTrigger: mockFireTableTrigger }))
 vi.mock('@/lib/table/constants', () => ({
   TABLE_LIMITS: { DELETE_PAGE_SIZE: 2 },
   USER_TABLE_ROWS_SQL_NAME: 'user_table_rows',
@@ -62,7 +65,13 @@ const UNLOCKED = {
   updateLocked: false,
   deleteLocked: false,
 }
-const table = { id: 'tbl_1', workspaceId: 'ws_1', schema: { columns: [] }, locks: UNLOCKED }
+const table = {
+  id: 'tbl_1',
+  name: 'Issues',
+  workspaceId: 'ws_1',
+  schema: { columns: [] },
+  locks: UNLOCKED,
+}
 const cutoff = new Date('2026-06-05T00:00:00Z')
 
 function basePayload(overrides = {}) {
@@ -77,7 +86,23 @@ describe('runTableDelete', () => {
     mockUpdateJobProgress.mockResolvedValue(true)
     mockMarkJobReady.mockResolvedValue(true)
     mockMarkJobFailed.mockResolvedValue(undefined)
-    mockDeletePageByIds.mockImplementation((_t, _w, ids: string[]) => Promise.resolve(ids.length))
+    mockDeletePageByIds.mockImplementation(
+      async (
+        _t,
+        _w,
+        ids: string[],
+        _proof,
+        _revalidate,
+        onDeleted?: (
+          rows: Array<{ id: string; data: Record<string, unknown> }>,
+          table?: typeof table
+        ) => void | Promise<void>
+      ) => {
+        const rows = ids.map((id) => ({ id, data: { title: id } }))
+        await onDeleted?.(rows)
+        return rows.length
+      }
+    )
     mockBuildFilterClause.mockReturnValue({})
   })
 
@@ -115,6 +140,7 @@ describe('runTableDelete', () => {
       'ws_1',
       ['a', 'b'],
       expect.anything(),
+      expect.any(Function),
       expect.any(Function)
     )
     expect(mockMarkJobCanceled).toHaveBeenCalledWith('tbl_1', 'job_1')
@@ -150,6 +176,7 @@ describe('runTableDelete', () => {
       'ws_1',
       ['a', 'b'],
       expect.anything(),
+      expect.any(Function),
       expect.any(Function)
     )
     expect(mockDeletePageByIds).toHaveBeenNthCalledWith(
@@ -158,17 +185,62 @@ describe('runTableDelete', () => {
       'ws_1',
       ['c'],
       expect.anything(),
+      expect.any(Function),
       expect.any(Function)
     )
     expect(mockMarkJobReady).toHaveBeenCalledWith('tbl_1', 'job_1')
     expect(mockAppendTableEvent).toHaveBeenCalledWith(
       expect.objectContaining({ kind: 'job', type: 'delete', status: 'ready', progress: 3 })
     )
+    expect(mockFireTableTrigger).toHaveBeenCalledTimes(2)
+    expect(mockFireTableTrigger).toHaveBeenNthCalledWith(
+      1,
+      'tbl_1',
+      'ws_1',
+      'Issues',
+      'delete',
+      [
+        { id: 'a', data: { title: 'a' } },
+        { id: 'b', data: { title: 'b' } },
+      ],
+      null,
+      table.schema,
+      expect.any(String)
+    )
     // The live grid must be told rows changed so deleted rows drop out of every open editor —
     // the `job` progress event only drives the delete meter, not the rows query.
     expect(mockSignalTableRowsChanged).toHaveBeenCalledWith('tbl_1')
   })
 
+  it('uses the table definition revalidated with each committed delete batch', async () => {
+    const renamedTable = {
+      ...table,
+      name: 'Renamed issues',
+      schema: { columns: [{ id: 'col-title', name: 'Renamed title', type: 'string' }] },
+    }
+    mockSelectRowIdPage.mockResolvedValueOnce(['a']).mockResolvedValueOnce([])
+    mockDeletePageByIds.mockImplementationOnce(
+      async (_t, _w, ids: string[], _proof, _revalidate, onDeleted) => {
+        const rows = ids.map((id) => ({ id, data: { 'col-title': id } }))
+        await onDeleted?.(rows, renamedTable)
+        return rows.length
+      }
+    )
+
+    await runTableDelete(basePayload())
+
+    expect(mockFireTableTrigger).toHaveBeenCalledWith(
+      renamedTable.id,
+      renamedTable.workspaceId,
+      renamedTable.name,
+      'delete',
+      [{ id: 'a', data: { 'col-title': 'a' } }],
+      null,
+      renamedTable.schema,
+      expect.any(String)
+    )
+  })
+
   it('stops once maxRows is reached and caps the final page fetch to the remaining budget', async () => {
     // budget 3 with page size 2: first page fills 2, the second is capped to the remaining 1.
     mockSelectRowIdPage.mockResolvedValueOnce(['a', 'b']).mockResolvedValueOnce(['c'])
@@ -196,6 +268,7 @@ describe('runTableDelete', () => {
       'ws_1',
       ['x'],
       expect.anything(),
+      expect.any(Function),
       expect.any(Function)
     )
     // Second page is queried after the last id of the first page (cursor advanced past 'keep').
diff --git a/apps/sim/lib/table/delete-runner.ts b/apps/sim/lib/table/delete-runner.ts
index a1c14302bed..6b010f02518 100644
--- a/apps/sim/lib/table/delete-runner.ts
+++ b/apps/sim/lib/table/delete-runner.ts
@@ -14,9 +14,10 @@ import {
 } from '@/lib/table/jobs/service'
 import { assertRowDelete, type MutationProof, TableLockedError } from '@/lib/table/mutation-locks'
 import type { DbTransaction } from '@/lib/table/planner'
-import { deletePageByIds, selectRowIdPage } from '@/lib/table/rows/ordering'
+import { type DeletedTableRow, deletePageByIds, selectRowIdPage } from '@/lib/table/rows/ordering'
 import { getTableById } from '@/lib/table/service'
 import { buildFilterClause } from '@/lib/table/sql'
+import { fireTableTrigger } from '@/lib/table/trigger'
 
 const logger = createLogger('TableDeleteRunner')
 
@@ -122,6 +123,22 @@ export async function runTableDelete(payload: TableDeletePayload): Promise<void>
     // an absent filter is still legitimate (delete-all is an explicit caller mode).
     if (filter && !filterClause) throw new Error('Filter is required for bulk delete')
     const excluded = new Set(excludeRowIds ?? [])
+    const dispatchDeleteTriggers = async (
+      rows: DeletedTableRow[],
+      committedTable?: TableDefinition
+    ) => {
+      const triggerTable = committedTable ?? table
+      await fireTableTrigger(
+        triggerTable.id,
+        triggerTable.workspaceId,
+        triggerTable.name,
+        'delete',
+        rows,
+        null,
+        triggerTable.schema,
+        requestId
+      )
+    }
 
     // Resume the persisted count: a retried attempt's earlier batches are already committed,
     // so starting at zero would overwrite cumulative progress with this attempt's smaller
@@ -170,7 +187,14 @@ export async function runTableDelete(payload: TableDeletePayload): Promise<void>
         // returns or throws. (An attempt that ends up committing nothing only over-refetches — harmless.)
         deletedAny = true
         try {
-          processed += await deletePageByIds(tableId, workspaceId, toDelete, pageProof, revalidate)
+          processed += await deletePageByIds(
+            tableId,
+            workspaceId,
+            toDelete,
+            pageProof,
+            revalidate,
+            dispatchDeleteTriggers
+          )
         } catch (err) {
           if (!(err instanceof TableLockedError)) throw err
           // A lock landed between batches. Batches already committed stay
diff --git a/apps/sim/lib/table/rows/__tests__/ordering-delete.test.ts b/apps/sim/lib/table/rows/__tests__/ordering-delete.test.ts
new file mode 100644
index 00000000000..eda638fcb47
--- /dev/null
+++ b/apps/sim/lib/table/rows/__tests__/ordering-delete.test.ts
@@ -0,0 +1,98 @@
+/**
+ * @vitest-environment node
+ */
+import { databaseMock } from '@sim/testing'
+import { beforeEach, describe, expect, it, vi } from 'vitest'
+import type { MutationProof } from '@/lib/table/mutation-locks'
+import type { DbTransaction } from '@/lib/table/planner'
+
+vi.mock('@/lib/table/constants', () => ({
+  getDeleteSnapshotBatchSize: () => 1,
+  TABLE_LIMITS: { UPDATE_BATCH_SIZE: 100 },
+}))
+vi.mock('@/lib/table/tx', () => ({ setTableTxTimeouts: vi.fn() }))
+
+import {
+  type DeletedRowsHandler,
+  deleteOrderedRowsByIds,
+  deletePageByIds,
+} from '@/lib/table/rows/ordering'
+
+const mockTransaction = databaseMock.db.transaction as ReturnType<typeof vi.fn>
+const proof = {} as MutationProof<'delete'>
+
+type DeleteRunner = (onDeleted: DeletedRowsHandler) => Promise<unknown>
+
+describe('ordered row delete trigger handoff', () => {
+  beforeEach(() => {
+    vi.clearAllMocks()
+  })
+
+  it.each([
+    [
+      'direct deletes',
+      (onDeleted: DeletedRowsHandler) =>
+        deleteOrderedRowsByIds({
+          tableId: 'table-1',
+          workspaceId: 'workspace-1',
+          rowIds: ['row-1', 'row-2'],
+          proof,
+          onDeleted,
+        }),
+    ],
+    [
+      'background delete pages',
+      (onDeleted: DeletedRowsHandler) =>
+        deletePageByIds('table-1', 'workspace-1', ['row-1', 'row-2'], proof, undefined, onDeleted),
+    ],
+  ])(
+    'runs %s handlers after commit and before the next batch',
+    async (_label, run: DeleteRunner) => {
+      const events: string[] = []
+      let batchIndex = 0
+      let releaseFirstHandler: (() => void) | undefined
+      const firstHandlerGate = new Promise<void>((resolve) => {
+        releaseFirstHandler = resolve
+      })
+      const trx = {
+        delete: () => ({
+          where: () => ({
+            returning: async () => {
+              const id = `row-${batchIndex + 1}`
+              batchIndex++
+              return [{ id, data: { title: id } }]
+            },
+          }),
+        }),
+      } as unknown as DbTransaction
+
+      mockTransaction.mockImplementation(
+        async (callback: (transaction: DbTransaction) => Promise<unknown>) => {
+          const result = await callback(trx)
+          events.push(`commit-${mockTransaction.mock.calls.length}`)
+          return result
+        }
+      )
+
+      const onDeleted = vi.fn(async (rows: Array<{ id: string }>) => {
+        events.push(`trigger-${rows[0]?.id}`)
+        if (rows[0]?.id === 'row-1') await firstHandlerGate
+      })
+      const pending = run(onDeleted)
+
+      await vi.waitFor(() => {
+        expect(events).toEqual(['commit-1', 'trigger-row-1'])
+      })
+      expect(mockTransaction).toHaveBeenCalledTimes(1)
+
+      releaseFirstHandler?.()
+      await pending
+
+      expect(events).toEqual(['commit-1', 'trigger-row-1', 'commit-2', 'trigger-row-2'])
+      expect(onDeleted.mock.calls.map(([rows]) => rows)).toEqual([
+        [{ id: 'row-1', data: { title: 'row-1' } }],
+        [{ id: 'row-2', data: { title: 'row-2' } }],
+      ])
+    }
+  )
+})
diff --git a/apps/sim/lib/table/rows/ordering.ts b/apps/sim/lib/table/rows/ordering.ts
index c776cc6e075..eba6c7b2bc4 100644
--- a/apps/sim/lib/table/rows/ordering.ts
+++ b/apps/sim/lib/table/rows/ordering.ts
@@ -10,7 +10,7 @@ import { db } from '@sim/db'
 import { userTableRows } from '@sim/db/schema'
 import { and, asc, desc, eq, gt, inArray, lt, lte, type SQL, sql } from 'drizzle-orm'
 import type { DbOrTx } from '@/lib/db/types'
-import { TABLE_LIMITS } from '@/lib/table/constants'
+import { getDeleteSnapshotBatchSize, TABLE_LIMITS } from '@/lib/table/constants'
 import type { MutationProof } from '@/lib/table/mutation-locks'
 import { keyBetween, nKeysBetween } from '@/lib/table/order-key'
 import { type DbExecutor, type DbTransaction, withSeqscanOff } from '@/lib/table/planner'
@@ -19,6 +19,16 @@ import { mutateTableRowsWithSecretProvenance } from '@/lib/table/rows/secret-pro
 import { setTableTxTimeouts } from '@/lib/table/tx'
 import type { RowData, TableDefinition, TableRowSecretProvenanceWrite } from '@/lib/table/types'
 
+export interface DeletedTableRow {
+  id: string
+  data: RowData
+}
+
+export type DeletedRowsHandler = (
+  rows: DeletedTableRow[],
+  table?: TableDefinition
+) => void | Promise<void>
+
 /**
  * Starting `position` for an append import — `max(position) + 1`, or 0 when empty. Read once,
  * unlocked, before streaming: the import worker is the table's sole writer, so it can assign
@@ -274,8 +284,8 @@ export async function insertOrderedRow(params: {
 
 /**
  * Deletes a single row by id in its own transaction. Deleting a row never changes
- * another row's `order_key`, so no positional reshift is needed. Returns `false`
- * when no row matched.
+ * another row's `order_key`, so no positional reshift is needed. Returns the
+ * deleted row snapshot, or `null` when no row matched.
  */
 export async function deleteOrderedRow(params: {
   tableId: string
@@ -283,7 +293,7 @@ export async function deleteOrderedRow(params: {
   workspaceId: string
   /** Proof the caller asserted the delete lock (see `mutation-locks.ts`). */
   proof: MutationProof<'delete'>
-}): Promise<boolean> {
+}): Promise<DeletedTableRow | null> {
   const { tableId, rowId, workspaceId } = params
   return db.transaction(async (trx) => {
     await setTableTxTimeouts(trx)
@@ -296,16 +306,18 @@ export async function deleteOrderedRow(params: {
           eq(userTableRows.workspaceId, workspaceId)
         )
       )
-      .returning({ id: userTableRows.id })
-    return Boolean(deleted)
+      .returning({ id: userTableRows.id, data: userTableRows.data })
+    return deleted ? { id: deleted.id, data: deleted.data as RowData } : null
   })
 }
 
 /**
- * Deletes the given row ids in batches within one transaction. Deletes leave
- * `order_key` untouched, so no positional recompaction is needed. Returns the
- * deleted row ids. The caller resolves which ids to delete (used by both
- * delete-by-ids and delete-by-filter).
+ * Deletes the given row ids in byte-bounded, independently committed batches.
+ * Deletes leave `order_key` untouched, so no positional recompaction is needed.
+ * The post-commit handler is awaited before the next batch so deleted JSON
+ * snapshots cannot accumulate in memory. Returns only the compact deleted ids;
+ * the caller resolves which ids to delete (used by both delete-by-ids and
+ * delete-by-filter).
  */
 export async function deleteOrderedRowsByIds(params: {
   tableId: string
@@ -313,15 +325,18 @@ export async function deleteOrderedRowsByIds(params: {
   rowIds: string[]
   /** Proof the caller asserted the delete lock (see `mutation-locks.ts`). */
   proof: MutationProof<'delete'>
-}): Promise<{ id: string }[]> {
-  const { tableId, workspaceId, rowIds } = params
+  /** Handles each bounded snapshot batch after its transaction commits. */
+  onDeleted?: DeletedRowsHandler
+}): Promise<string[]> {
+  const { tableId, workspaceId, rowIds, onDeleted } = params
   if (rowIds.length === 0) return []
-  return db.transaction(async (trx) => {
-    await setTableTxTimeouts(trx, { statementMs: 60_000 })
-    const deleted: { id: string }[] = []
-    for (let i = 0; i < rowIds.length; i += TABLE_LIMITS.DELETE_BATCH_SIZE) {
-      const batch = rowIds.slice(i, i + TABLE_LIMITS.DELETE_BATCH_SIZE)
-      const rows = await trx
+  const batchSize = getDeleteSnapshotBatchSize()
+  const deletedIds: string[] = []
+  for (let i = 0; i < rowIds.length; i += batchSize) {
+    const batch = rowIds.slice(i, i + batchSize)
+    const rows = await db.transaction(async (trx) => {
+      await setTableTxTimeouts(trx, { statementMs: 60_000 })
+      return trx
         .delete(userTableRows)
         .where(
           and(
@@ -330,11 +345,13 @@ export async function deleteOrderedRowsByIds(params: {
             inArray(userTableRows.id, batch)
           )
         )
-        .returning({ id: userTableRows.id })
-      deleted.push(...rows)
-    }
-    return deleted
-  })
+        .returning({ id: userTableRows.id, data: userTableRows.data })
+    })
+    const deletedRows = rows.map((row) => ({ id: row.id, data: row.data as RowData }))
+    deletedIds.push(...deletedRows.map((row) => row.id))
+    await onDeleted?.(deletedRows)
+  }
+  return deletedIds
 }
 
 /**
@@ -467,15 +484,18 @@ export async function deletePageByIds(
   /** Proof the caller asserted the delete lock (see `mutation-locks.ts`). */
   _proof: MutationProof<'delete'>,
   /** Re-asserts the lock inside each batch transaction. See {@link guardBatch}. */
-  revalidate?: MutationRevalidator
+  revalidate?: MutationRevalidator,
+  /** Called after each batch commits, with snapshots suitable for delete triggers. */
+  onDeleted?: DeletedRowsHandler
 ): Promise<number> {
   let deleted = 0
-  for (let i = 0; i < rowIds.length; i += TABLE_LIMITS.DELETE_BATCH_SIZE) {
-    const batch = rowIds.slice(i, i + TABLE_LIMITS.DELETE_BATCH_SIZE)
-    const rows = await db.transaction(async (trx) => {
+  const batchSize = getDeleteSnapshotBatchSize()
+  for (let i = 0; i < rowIds.length; i += batchSize) {
+    const batch = rowIds.slice(i, i + batchSize)
+    const { rows, table } = await db.transaction(async (trx) => {
       await setTableTxTimeouts(trx, { statementMs: 60_000 })
-      await guardBatch(trx, tableId, revalidate)
-      return trx
+      const table = await guardBatch(trx, tableId, revalidate)
+      const rows = await trx
         .delete(userTableRows)
         .where(
           and(
@@ -484,9 +504,12 @@ export async function deletePageByIds(
             inArray(userTableRows.id, batch)
           )
         )
-        .returning({ id: userTableRows.id })
+        .returning({ id: userTableRows.id, data: userTableRows.data })
+      return { rows, table }
     })
-    deleted += rows.length
+    const deletedRows = rows.map((row) => ({ id: row.id, data: row.data as RowData }))
+    deleted += deletedRows.length
+    await onDeleted?.(deletedRows, table)
   }
   return deleted
 }
diff --git a/apps/sim/lib/table/rows/service.ts b/apps/sim/lib/table/rows/service.ts
index f8933c05cdf..d322382ea9c 100644
--- a/apps/sim/lib/table/rows/service.ts
+++ b/apps/sim/lib/table/rows/service.ts
@@ -51,6 +51,7 @@ import {
 } from '@/lib/table/rows/executions'
 import {
   acquireRowOrderLock,
+  type DeletedTableRow,
   deleteOrderedRow,
   deleteOrderedRowsByIds,
   insertOrderedRow,
@@ -110,6 +111,24 @@ import { cancelWorkflowGroupRuns, runWorkflowColumn } from '@/lib/table/workflow
 
 const logger = createLogger('TableRowsService')
 
+async function dispatchDeleteTriggers(
+  table: TableDefinition,
+  deletedRows: DeletedTableRow[],
+  requestId: string
+): Promise<void> {
+  if (deletedRows.length === 0) return
+  await fireTableTrigger(
+    table.id,
+    table.workspaceId,
+    table.name,
+    'delete',
+    deletedRows,
+    null,
+    table.schema,
+    requestId
+  )
+}
+
 /**
  * Inserts a single row into a table.
  *
@@ -207,6 +226,7 @@ export async function insertRow(
 
   void fireTableTrigger(
     data.tableId,
+    table.workspaceId,
     table.name,
     'insert',
     [insertedRow],
@@ -383,7 +403,16 @@ export function dispatchAfterBatchInsert(
   requestId: string,
   actorUserId?: string | null
 ): void {
-  void fireTableTrigger(table.id, table.name, 'insert', result, null, table.schema, requestId)
+  void fireTableTrigger(
+    table.id,
+    table.workspaceId,
+    table.name,
+    'insert',
+    result,
+    null,
+    table.schema,
+    requestId
+  )
   // Scope to the newly-inserted row ids so the dispatcher doesn't walk every
   // row in the table. After the sidecar migration, all existing rows have
   // zero entries → `mode:'new'`'s `NOT EXISTS` filter would otherwise include
@@ -865,6 +894,7 @@ export async function upsertRow(
     })
     void fireTableTrigger(
       data.tableId,
+      table.workspaceId,
       table.name,
       'insert',
       [result.row],
@@ -876,6 +906,7 @@ export async function upsertRow(
     const oldRows = new Map([[result.row.id, result.previousData]])
     void fireTableTrigger(
       data.tableId,
+      table.workspaceId,
       table.name,
       'update',
       [result.row],
@@ -1801,6 +1832,7 @@ export async function updateRow(
   const oldRows = new Map([[data.rowId, existingRow.data as RowData]])
   void fireTableTrigger(
     data.tableId,
+    table.workspaceId,
     table.name,
     'update',
     [updatedRow],
@@ -1886,6 +1918,7 @@ export async function deleteRow(
   if (!deleted) throw new OrchestrationError('not_found', 'Row not found')
 
   logger.info(`[${requestId}] Deleted row ${rowId} from table ${table.id}`)
+  await dispatchDeleteTriggers(table, [deleted], requestId)
 }
 
 type BulkUpdateMatch = { id: string; data: RowData }
@@ -2061,6 +2094,7 @@ function dispatchBulkUpdateEffects(
   }))
   void fireTableTrigger(
     table.id,
+    table.workspaceId,
     table.name,
     'update',
     updatedRows,
@@ -2483,6 +2517,7 @@ export async function batchUpdateRows(
   if (updatedRowsForTrigger.length > 0) {
     void fireTableTrigger(
       data.tableId,
+      table.workspaceId,
       table.name,
       'update',
       updatedRowsForTrigger,
@@ -2573,7 +2608,7 @@ export async function deleteRowsByFilter(
   )
 
   const limit = data.limit
-  const deletedRows: { id: string }[] = []
+  const deletedRowIds: string[] = []
   if (limit === undefined) {
     const cutoff = new Date()
     let afterId: string | undefined
@@ -2589,14 +2624,14 @@ export async function deleteRowsByFilter(
       if (page.length === 0) break
       const nextAfterId = page[page.length - 1]
       for (let index = 0; index < page.length; index += TABLE_LIMITS.DELETE_BATCH_SIZE) {
-        deletedRows.push(
-          ...(await deleteOrderedRowsByIds({
-            tableId: table.id,
-            workspaceId: table.workspaceId,
-            rowIds: page.slice(index, index + TABLE_LIMITS.DELETE_BATCH_SIZE),
-            proof,
-          }))
-        )
+        const deletedIds = await deleteOrderedRowsByIds({
+          tableId: table.id,
+          workspaceId: table.workspaceId,
+          rowIds: page.slice(index, index + TABLE_LIMITS.DELETE_BATCH_SIZE),
+          proof,
+          onDeleted: (rows) => dispatchDeleteTriggers(table, rows, requestId),
+        })
+        deletedRowIds.push(...deletedIds)
       }
       afterId = nextAfterId
       if (page.length < TABLE_LIMITS.DELETE_PAGE_SIZE) break
@@ -2612,19 +2647,18 @@ export async function deleteRowsByFilter(
     )
     const rowIds = matchingRows.map((row) => row.id)
     if (rowIds.length > 0) {
-      deletedRows.push(
-        ...(await deleteOrderedRowsByIds({
-          tableId: table.id,
-          workspaceId: table.workspaceId,
-          rowIds,
-          proof,
-        }))
-      )
+      const deletedIds = await deleteOrderedRowsByIds({
+        tableId: table.id,
+        workspaceId: table.workspaceId,
+        rowIds,
+        proof,
+        onDeleted: (rows) => dispatchDeleteTriggers(table, rows, requestId),
+      })
+      deletedRowIds.push(...deletedIds)
     }
   }
 
-  if (deletedRows.length === 0) return { affectedCount: 0, affectedRowIds: [] }
-  const deletedRowIds = deletedRows.map((row) => row.id)
+  if (deletedRowIds.length === 0) return { affectedCount: 0, affectedRowIds: [] }
 
   logger.info(`[${requestId}] Deleted ${deletedRowIds.length} rows from table ${table.id}`)
 
@@ -2650,19 +2684,18 @@ export async function deleteRowsByIds(
 
   const uniqueRequestedRowIds = Array.from(new Set(data.rowIds))
 
-  const deletedRows = await deleteOrderedRowsByIds({
+  const deletedIds = await deleteOrderedRowsByIds({
     tableId: data.tableId,
     workspaceId: data.workspaceId,
     rowIds: uniqueRequestedRowIds,
     proof,
+    onDeleted: (rows) => dispatchDeleteTriggers(table, rows, requestId),
   })
 
-  const deletedIds = deletedRows.map((r) => r.id)
   const deletedIdSet = new Set(deletedIds)
   const missingRowIds = uniqueRequestedRowIds.filter((id) => !deletedIdSet.has(id))
 
   logger.info(`[${requestId}] Deleted ${deletedIds.length} rows by ID from table ${data.tableId}`)
-
   return {
     deletedCount: deletedIds.length,
     deletedRowIds: deletedIds,
diff --git a/apps/sim/lib/table/trigger.ts b/apps/sim/lib/table/trigger.ts
index a08ebc093a0..53623598ac0 100644
--- a/apps/sim/lib/table/trigger.ts
+++ b/apps/sim/lib/table/trigger.ts
@@ -1,7 +1,7 @@
 /**
  * Direct trigger firing for table row events.
  *
- * When rows are inserted or updated in a table, this module looks up any
+ * When rows are inserted, updated, or deleted in a table, this module looks up any
  * active webhook triggers watching that table and fires workflow executions
  * immediately - no polling or cron involved.
  */
@@ -15,7 +15,8 @@ import { readCanonicalTriggerValue } from '@/lib/webhooks/polling/canonical'
 
 const logger = createLogger('TableTrigger')
 
-type EventType = 'insert' | 'update'
+type EventType = 'insert' | 'update' | 'delete'
+type TableTriggerRow = Pick<TableRow, 'id' | 'data'>
 
 interface TableTriggerPayload {
   row: Record<string, unknown> | null
@@ -44,16 +45,17 @@ interface WebhookConfig {
  * This is fire-and-forget - errors are logged but never thrown.
  * Call with `void fireTableTrigger(...)` to avoid blocking the caller.
  *
- * @param eventType - 'insert' for new rows, 'update' for changed rows
- * @param oldRows - Map of row ID to previous data. Pass null for inserts.
+ * @param workspaceId - Canonical workspace that owns the mutated table.
+ * @param eventType - The committed row mutation that should trigger workflows.
+ * @param rows - Committed row snapshots; only the ID and data are needed, including for deletes.
+ * @param oldRows - Map of row ID to previous data. Pass null for inserts and deletes.
  */
 export async function fireTableTrigger(
   tableId: string,
+  workspaceId: string,
   tableName: string,
   eventType: EventType,
-  // Accepts a row without its executions sidecar: the payload projects id and
-  // data only, and the upsert path deliberately does not load one.
-  rows: Array<Omit<TableRow, 'executions'>>,
+  rows: TableTriggerRow[],
   oldRows: Map<string, RowData> | null,
   schema: TableSchema,
   requestId: string
@@ -75,6 +77,7 @@ export async function fireTableTrigger(
 
     // Filter to webhooks watching this table with a matching event type
     const matching = webhooks.filter((entry) => {
+      if (entry.workflow.workspaceId !== workspaceId) return false
       const config = entry.webhook.providerConfig as WebhookConfig | null
       // Canonical key `tableId` first; `tableSelector`/`manualTableId` are a transitional
       // basic-first fallback for configs deployed before the canonical key was written.
@@ -103,7 +106,7 @@ export async function fireTableTrigger(
       const includeHeaders = config?.includeHeaders !== false
 
       for (const row of rows) {
-        const previousIdData = oldRows?.get(row.id) ?? null
+        const previousIdData = eventType === 'delete' ? row.data : (oldRows?.get(row.id) ?? null)
         const rawRow = toNamedRow(row.data)
         const previousRow = previousIdData ? toNamedRow(previousIdData) : null
         const changedColumns = previousIdData
diff --git a/apps/sim/triggers/table/poller.test.ts b/apps/sim/triggers/table/poller.test.ts
new file mode 100644
index 00000000000..1e421ccf1ad
--- /dev/null
+++ b/apps/sim/triggers/table/poller.test.ts
@@ -0,0 +1,26 @@
+/**
+ * @vitest-environment node
+ */
+import { describe, expect, it } from 'vitest'
+import { tableNewRowTrigger } from '@/triggers/table/poller'
+
+describe('table trigger events', () => {
+  it('offers insert, update, and delete row events', () => {
+    const eventType = tableNewRowTrigger.subBlocks.find((subBlock) => subBlock.id === 'eventType')
+    const options = eventType?.options as Array<{ id: string; label: string }> | undefined
+
+    expect(options).toEqual([
+      { id: 'insert', label: 'Row Inserted' },
+      { id: 'update', label: 'Row Updated' },
+      { id: 'delete', label: 'Row Deleted' },
+    ])
+  })
+
+  it('keeps watched-column filtering specific to updates', () => {
+    const watchColumns = tableNewRowTrigger.subBlocks.find(
+      (subBlock) => subBlock.id === 'watchColumns'
+    )
+
+    expect(watchColumns?.condition).toEqual({ field: 'eventType', value: 'update' })
+  })
+})
diff --git a/apps/sim/triggers/table/poller.ts b/apps/sim/triggers/table/poller.ts
index 99b9ed88c46..03b06510271 100644
--- a/apps/sim/triggers/table/poller.ts
+++ b/apps/sim/triggers/table/poller.ts
@@ -5,7 +5,7 @@ export const tableNewRowTrigger: TriggerConfig = {
   id: 'table_new_row',
   name: 'Table Trigger',
   provider: 'table',
-  description: 'Triggers when rows are inserted or updated in a table',
+  description: 'Triggers when rows are inserted, updated, or deleted in a table',
   version: '1.0.0',
   icon: Table,
 
@@ -37,6 +37,7 @@ export const tableNewRowTrigger: TriggerConfig = {
       options: [
         { id: 'insert', label: 'Row Inserted' },
         { id: 'update', label: 'Row Updated' },
+        { id: 'delete', label: 'Row Deleted' },
       ],
       defaultValue: 'insert',
       description: 'The type of event to trigger on.',
@@ -73,7 +74,7 @@ export const tableNewRowTrigger: TriggerConfig = {
       type: 'text',
       defaultValue: [
         'Select the table to monitor',
-        'Choose whether to trigger on row inserts or updates',
+        'Choose whether to trigger on row inserts, updates, or deletes',
         'For updates, optionally select specific columns to watch',
         'The workflow will trigger automatically when the event occurs',
       ]
@@ -97,11 +98,11 @@ export const tableNewRowTrigger: TriggerConfig = {
     },
     previousRow: {
       type: 'json',
-      description: 'Previous row data before the update (null for inserts)',
+      description: 'Previous row data before an update or deletion (null for inserts)',
     },
     changedColumns: {
       type: 'json',
-      description: 'List of column names that changed (empty for inserts)',
+      description: 'List of column names that changed (empty for inserts and deletes)',
     },
     rowId: {
       type: 'string',