@@ -15,13 +15,15 @@ const {
   mockSignalTableRowsChanged,
   mockTask,
   mockWithLockedTable,
+  mockFireTableTrigger,
 } = vi.hoisted(() => ({
   mockDeleteExecute: vi.fn(),
   mockListExecute: vi.fn(),
   mockIsTableRowTtlEnabled: vi.fn(),
   mockSignalTableRowsChanged: vi.fn(),
   mockTask: vi.fn((config: unknown) => config),
   mockWithLockedTable: vi.fn(),
+  mockFireTableTrigger: vi.fn(),
 }))
 
 vi.mock('@sim/db', () => ({
@@ -34,18 +36,34 @@ 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'
 
 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 +83,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 +103,24 @@ 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.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 +137,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 +148,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 +204,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 +234,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 []
         }),
       })
     })
 
@@ -8,8 +8,11 @@ import { getColumnId } from '@/lib/table/column-keys'
 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')
@@ -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): 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 > TTL_CLEANUP_BATCH_SIZE) {
     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),
   }
 }
 
@@ -123,12 +130,8 @@ async function deleteExpiredTableRowBatch(
   columnKey: string,
   nowEpochSeconds: 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
@@ -148,18 +151,14 @@ async function deleteExpiredTableRowBatch(
       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)
 }
@@ -170,9 +169,9 @@ async function deleteExpiredRowsForTable(
   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) {
@@ -193,10 +192,27 @@ async function deleteExpiredRowsForTable(
           nowEpochSeconds,
           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) {
+      void fireTableTrigger(
+        ref.id,
+        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 }
 
@@ -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,55 @@ 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.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.name,
+      'delete',
+      [
+        { id: 'row-1', data: { name: 'Ada' } },
+        { id: 'row-2', data: { name: 'Grace' } },
+      ],
+      null,
+      TABLE.schema,
+      'req-delete-many'
+    )
+  })
+})
+
 describe('bulk update/delete limited-subset ordering', () => {
   beforeEach(() => {
     vi.clearAllMocks()