diff --git a/apps/sim/lib/auth/admin-removal-stops.test.ts b/apps/sim/lib/auth/admin-removal-stops.test.ts
new file mode 100644
index 00000000000..022d2f04f8f
--- /dev/null
+++ b/apps/sim/lib/auth/admin-removal-stops.test.ts
@@ -0,0 +1,59 @@
+/**
+ * The `/admin/remove-user` hooks hand a plan from `before` to `after` across a non-transactional
+ * delete. Failure modes:
+ * - the outbox write fails after the delete committed → the in-flight runs must still be signalled
+ * - `after` runs for a different delete than the one `before` read → nothing is stopped
+ */
+import { workflowExecutionLogs, workspace } from '@sim/db/schema'
+import { dbChainMockFns, queueTableRows, resetDbChainMock } from '@sim/testing/mocks/database.mock'
+import { redisConfigMockFns } from '@sim/testing/mocks/redis-config.mock'
+import { beforeEach, describe, expect, it, vi } from 'vitest'
+import { holdAdminRemovalStops, stopAdminRemovedRuns } from '@/lib/auth/admin-removal-stops'
+
+const mockEval = vi.fn()
+
+function queueRunningExecution(executionId: string) {
+  queueTableRows(workspace, [])
+  queueTableRows(workflowExecutionLogs, [
+    { executionId, executionDeadlineAt: null, workspaceId: 'workspace-elsewhere' },
+  ])
+  queueTableRows(workflowExecutionLogs, [])
+}
+
+describe('admin user removal stops', () => {
+  beforeEach(() => {
+    resetDbChainMock()
+    mockEval.mockReset().mockResolvedValue(1)
+    redisConfigMockFns.mockGetRedisClient.mockReturnValue({ eval: mockEval })
+  })
+
+  it('signals the removed user’s runs even when their stops cannot be recorded', async () => {
+    const user = { id: 'user-1' }
+    queueRunningExecution('execution-1')
+    await holdAdminRemovalStops(user)
+    dbChainMockFns.insert.mockImplementationOnce(() => {
+      throw new Error('outbox unavailable')
+    })
+
+    await expect(stopAdminRemovedRuns(user)).resolves.toBeUndefined()
+
+    expect(mockEval).toHaveBeenCalledWith(
+      expect.any(String),
+      1,
+      'execution:cancel:execution-1',
+      expect.anything(),
+      expect.anything(),
+      expect.anything(),
+      expect.anything()
+    )
+  })
+
+  it('stops nothing for a delete whose runs it did not read', async () => {
+    queueRunningExecution('execution-2')
+    await holdAdminRemovalStops({ id: 'user-2' })
+
+    await stopAdminRemovedRuns({ id: 'user-2' })
+
+    expect(mockEval).not.toHaveBeenCalled()
+  })
+})
diff --git a/apps/sim/lib/auth/admin-removal-stops.ts b/apps/sim/lib/auth/admin-removal-stops.ts
new file mode 100644
index 00000000000..3683a87a3e8
--- /dev/null
+++ b/apps/sim/lib/auth/admin-removal-stops.ts
@@ -0,0 +1,55 @@
+import { db } from '@sim/db'
+import { createLogger } from '@sim/logger'
+import { getErrorMessage } from '@sim/utils/errors'
+import {
+  collectAccountRemovalStops,
+  type ExecutionStopPlan,
+  enqueueExecutionStops,
+  signalExecutionStops,
+} from '@/lib/execution/removal-stop-records'
+
+const logger = createLogger('AdminRemovalStops')
+
+/**
+ * Plans read by {@link holdAdminRemovalStops} and acted on by {@link stopAdminRemovedRuns},
+ * keyed by the user entity Better Auth passes to both hooks of one delete. A delete that fails
+ * never reaches `after`, so its plan is dropped with the entity.
+ */
+const pendingStops = new WeakMap<object, ExecutionStopPlan>()
+
+/**
+ * The user delete's `before` hook: `/admin/remove-user` deletes the user row outside any
+ * transaction, and the cascade erases the rows naming its runs, so they are read now. A failed
+ * read lets the removal proceed as before rather than blocking an admin.
+ */
+export async function holdAdminRemovalStops(user: { id: string }): Promise<void> {
+  try {
+    pendingStops.set(user, await collectAccountRemovalStops(db, user.id))
+  } catch (error) {
+    logger.error('Failed to read the runs a user removal must stop', {
+      userId: user.id,
+      error: getErrorMessage(error),
+    })
+  }
+}
+
+/**
+ * The user delete's `after` hook, once the delete has committed: signals the in-flight runs at
+ * once, whether or not their stops could be recorded, and records the stops for the outbox
+ * worker's queue sweep and retries. Never throws.
+ */
+export async function stopAdminRemovedRuns(user: { id: string }): Promise<void> {
+  const plan = pendingStops.get(user)
+  pendingStops.delete(user)
+  if (!plan) return
+  try {
+    await enqueueExecutionStops(db, plan)
+  } catch (error) {
+    logger.error('Failed to record the stops of a removed user’s runs', {
+      userId: user.id,
+      error: getErrorMessage(error),
+    })
+  } finally {
+    await signalExecutionStops(plan.executions)
+  }
+}
diff --git a/apps/sim/lib/auth/auth.ts b/apps/sim/lib/auth/auth.ts
index 4efd9cf70c5..110d53f4060 100644
--- a/apps/sim/lib/auth/auth.ts
+++ b/apps/sim/lib/auth/auth.ts
@@ -38,6 +38,7 @@ import { readAttributionProperties } from '@/lib/analytics/attribution'
 import { FREEBUFF_CLICK_ID_COOKIE } from '@/lib/analytics/freebuff'
 import { reportFreebuffConversion } from '@/lib/analytics/freebuff.server'
 import { getAccessControlConfig, isEmailBlockedByAccessControl } from '@/lib/auth/access-control'
+import { holdAdminRemovalStops, stopAdminRemovedRuns } from '@/lib/auth/admin-removal-stops'
 import { createAnonymousSession, ensureAnonymousUserExists } from '@/lib/auth/anonymous'
 import { buildConnectorProviders } from '@/lib/auth/connectors/providers'
 import {
@@ -469,11 +470,15 @@ export const auth = betterAuth({
       },
       /**
        * The admin plugin's `/admin/remove-user` deletes the user row through Better Auth, so its
-       * usage rows cascade away without `deleteUserAccount`. Better Auth runs this once the delete
-       * commits, which is when the stored reporting prefixes that counted them must be discarded.
+       * usage rows cascade away without `deleteUserAccount`. Better Auth runs `after` once the
+       * delete commits, which is when the stored reporting prefixes that counted them must be
+       * discarded, and when the runs the cascade orphaned are stopped: in-flight ones at once,
+       * queued ones by the outbox worker, since this module cannot load that delivery.
        */
       delete: {
-        after: async () => {
+        before: holdAdminRemovalStops,
+        after: async (user) => {
+          await stopAdminRemovedRuns(user)
           await invalidateReportingUsagePrefixes()
         },
       },
diff --git a/apps/sim/lib/core/async-jobs/backends/database.test.ts b/apps/sim/lib/core/async-jobs/backends/database.test.ts
index 0046926f940..5b4cce31ca9 100644
--- a/apps/sim/lib/core/async-jobs/backends/database.test.ts
+++ b/apps/sim/lib/core/async-jobs/backends/database.test.ts
@@ -1,4 +1,10 @@
-import { dbChainMockFns, resetDbChainMock } from '@sim/testing/mocks/database.mock'
+import { asyncJobs } from '@sim/db'
+import {
+  dbChainMockFns,
+  flattenMockConditions,
+  queueTableRows,
+  resetDbChainMock,
+} from '@sim/testing/mocks/database.mock'
 import { sleep } from '@sim/utils/helpers'
 import { beforeEach, describe, expect, it, vi } from 'vitest'
 
@@ -503,3 +509,69 @@ describe('DatabaseJobQueue a workflow job that completed with a recorded failure
     })
   })
 })
+
+describe('DatabaseJobQueue workspace cancellation', () => {
+  beforeEach(() => {
+    resetDbChainMock()
+  })
+
+  it('cancels only workflow job types, so the workspace’s other jobs keep running', async () => {
+    await new DatabaseJobQueue().cancelByWorkspace('workspace-1', { maxRuns: 1000 })
+
+    const typeFilter = flattenMockConditions(dbChainMockFns.where.mock.calls[0][0]).find(
+      (node) => node.type === 'inArray' && node.column === asyncJobs.type
+    )
+    expect(typeFilter?.values).toEqual([
+      'workflow-execution',
+      'schedule-execution',
+      'webhook-execution',
+      'resume-execution',
+      'workflow-group-cell',
+    ])
+  })
+
+  const ids = (count: number, offset = 0) =>
+    Array.from({ length: count }, (_, index) => ({ id: `job-${offset + index}` }))
+
+  it('keeps cancelling in batches until a short batch shows none are left', async () => {
+    dbChainMockFns.returning
+      .mockResolvedValueOnce(ids(100))
+      .mockResolvedValueOnce(ids(100, 100))
+      .mockResolvedValueOnce(ids(30, 200))
+
+    await expect(
+      new DatabaseJobQueue().cancelByWorkspace('workspace-1', { maxRuns: 1000 })
+    ).resolves.toEqual({ cancelledJobs: 230, complete: true })
+  })
+
+  it('reports a short batch incomplete while matching jobs remain', async () => {
+    dbChainMockFns.returning.mockResolvedValueOnce(ids(99))
+    queueTableRows(asyncJobs, [{ id: 'job-100' }])
+
+    await expect(
+      new DatabaseJobQueue().cancelByWorkspace('workspace-1', { maxRuns: 1000 })
+    ).resolves.toEqual({ cancelledJobs: 99, complete: false })
+  })
+
+  it('stops at its run cap and reports the sweep incomplete', async () => {
+    dbChainMockFns.returning.mockResolvedValueOnce(ids(100)).mockResolvedValueOnce(ids(50, 100))
+
+    await expect(
+      new DatabaseJobQueue().cancelByWorkspace('workspace-1', { maxRuns: 150 })
+    ).resolves.toEqual({ cancelledJobs: 150, complete: false })
+    expect(dbChainMockFns.returning).toHaveBeenCalledTimes(2)
+  })
+
+  it('cancels nothing once its caller’s time is up', async () => {
+    const controller = new AbortController()
+    controller.abort()
+
+    await expect(
+      new DatabaseJobQueue().cancelByWorkspace('workspace-1', {
+        maxRuns: 1000,
+        signal: controller.signal,
+      })
+    ).resolves.toEqual({ cancelledJobs: 0, complete: false })
+    expect(dbChainMockFns.update).not.toHaveBeenCalled()
+  })
+})
diff --git a/apps/sim/lib/core/async-jobs/backends/database.ts b/apps/sim/lib/core/async-jobs/backends/database.ts
index 05c6b9bb01f..972a220d5db 100644
--- a/apps/sim/lib/core/async-jobs/backends/database.ts
+++ b/apps/sim/lib/core/async-jobs/backends/database.ts
@@ -18,10 +18,14 @@ import {
   type JobStatus,
   type JobType,
   validateMaxDurationSeconds,
+  WORKSPACE_EXECUTION_JOB_TYPES,
+  type WorkspaceJobCancellationOptions,
+  type WorkspaceJobCancellationResult,
 } from '@/lib/core/async-jobs/types'
 import { recordExecutionCancellationBackendResult } from '@/lib/core/execution-limits/metrics'
 
 const logger = createLogger('DatabaseJobQueue')
+const WORKSPACE_CANCELLATION_BATCH_SIZE = 100
 
 type AsyncJobRow = typeof asyncJobs.$inferSelect
 type Runner = NonNullable<EnqueueOptions['runner']>
@@ -51,6 +55,15 @@ function rowToJob(row: AsyncJobRow): Job {
 const inlineAbortControllers = new Map<string, AbortController>()
 const inlineExecutionControllers = new Map<string, Map<AbortController, JobType>>()
 
+/** Aborts the runner this process holds for the job, if any. */
+function abortInlineJob(jobId: string): boolean {
+  const controller = inlineAbortControllers.get(jobId)
+  if (!controller) return false
+  controller.abort(new DOMException('user', 'AbortError'))
+  inlineAbortControllers.delete(jobId)
+  return true
+}
+
 /**
  * Per-cancel-key abort controllers for the `batchEnqueueAndWait` direct-call
  * path. Distinct from `inlineAbortControllers` (which keys by jobId) — this
@@ -420,13 +433,7 @@ export class DatabaseJobQueue implements JobQueueBackend {
     // Abort any in-process inline execution first so the running workflow
     // observes the signal and stops mid-flight. Then mark the row cancelled so
     // any future poller skips it without treating a user action as a failure.
-    const controller = inlineAbortControllers.get(jobId)
-    let aborted = false
-    if (controller) {
-      controller.abort(new DOMException('user', 'AbortError'))
-      inlineAbortControllers.delete(jobId)
-      aborted = true
-    }
+    const aborted = abortInlineJob(jobId)
 
     const now = new Date()
     await db
@@ -517,6 +524,65 @@ export class DatabaseJobQueue implements JobQueueBackend {
     }
   }
 
+  /**
+   * Cancels the workspace's pending and processing rows, so no process can claim them, and
+   * aborts the runners this process holds for them. A runner another process holds stops through
+   * its execution's cancellation signal instead. Batches run through `batchEnqueueAndWait` keep
+   * no row and are left to that signal and their dispatch's own cancellation. Works in batches
+   * up to `maxRuns`, checking `signal` between them, like the Trigger.dev sweep.
+   */
+  async cancelByWorkspace(
+    workspaceId: string,
+    options: WorkspaceJobCancellationOptions
+  ): Promise<WorkspaceJobCancellationResult> {
+    const active = and(
+      inArray(asyncJobs.status, [JOB_STATUS.PENDING, JOB_STATUS.PROCESSING]),
+      inArray(asyncJobs.type, [...WORKSPACE_EXECUTION_JOB_TYPES]),
+      sql`${asyncJobs.metadata}->>'workspaceId' = ${workspaceId}`
+    )
+    let cancelledJobs = 0
+    while (cancelledJobs < options.maxRuns && !options.signal?.aborted) {
+      const batchSize = Math.min(WORKSPACE_CANCELLATION_BATCH_SIZE, options.maxRuns - cancelledJobs)
+      const now = new Date()
+      const cancelled = await db
+        .update(asyncJobs)
+        .set({ status: JOB_STATUS.CANCELLED, completedAt: now, error: 'Cancelled', updatedAt: now })
+        .where(
+          and(
+            active,
+            inArray(
+              asyncJobs.id,
+              db.select({ id: asyncJobs.id }).from(asyncJobs).where(active).limit(batchSize)
+            )
+          )
+        )
+        .returning({ id: asyncJobs.id })
+
+      for (const { id } of cancelled) abortInlineJob(id)
+      cancelledJobs += cancelled.length
+      if (cancelled.length < batchSize) {
+        /**
+         * A short batch is not proof the sweep is done: a selected job that finished before the
+         * update reached it shrinks the batch while later jobs may still match.
+         */
+        const [remaining] = await db
+          .select({ id: asyncJobs.id })
+          .from(asyncJobs)
+          .where(active)
+          .limit(1)
+        logger.info('Cancelled database jobs for workspace', {
+          workspaceId,
+          cancelledJobs,
+          complete: !remaining,
+        })
+        return { cancelledJobs, complete: !remaining }
+      }
+    }
+
+    logger.info('Paused database job sweep for workspace', { workspaceId, cancelledJobs })
+    return { cancelledJobs, complete: false }
+  }
+
   cancelByKey(cancelKey: string): boolean {
     const controller = inlineCancelKeyControllers.get(cancelKey)
     if (!controller) return false
diff --git a/apps/sim/lib/core/async-jobs/backends/trigger-dev.test.ts b/apps/sim/lib/core/async-jobs/backends/trigger-dev.test.ts
index 92df6bc14f0..c11d8276713 100644
--- a/apps/sim/lib/core/async-jobs/backends/trigger-dev.test.ts
+++ b/apps/sim/lib/core/async-jobs/backends/trigger-dev.test.ts
@@ -743,3 +743,246 @@ describe('TriggerDevJobQueue cancellation', () => {
     expect(maxActiveCancellations).toBe(10)
   })
 })
+
+describe('TriggerDevJobQueue workspace cancellation', () => {
+  interface FakeRun {
+    id: string
+    taskIdentifier: string
+    tags: string[]
+    status: string
+  }
+
+  /** Stands in for Trigger.dev: runs carry the tags enqueue sent, and listing honors its filters. */
+  function installFakeTrigger() {
+    const runs: FakeRun[] = []
+    mockTrigger.mockImplementation(
+      async (taskIdentifier: string, _payload?: unknown, options?: unknown) => {
+        const run = {
+          id: `run-${runs.length}`,
+          taskIdentifier,
+          tags: (options as { tags?: string[] } | undefined)?.tags ?? [],
+          status: 'QUEUED',
+        }
+        runs.push(run)
+        return { id: run.id }
+      }
+    )
+    mockList.mockImplementation((query?: unknown) => {
+      const filter = query as { tag: string; taskIdentifier: string[]; status: string[] }
+      const matches = runs.filter(
+        (run) =>
+          run.tags.includes(filter.tag) &&
+          filter.taskIdentifier.includes(run.taskIdentifier) &&
+          filter.status.includes(run.status)
+      )
+      return createListPage(matches)
+    })
+    mockCancel.mockImplementation(async (runId: string) => {
+      const run = runs.find((candidate) => candidate.id === runId)
+      if (run) run.status = 'CANCELED'
+    })
+    return runs
+  }
+
+  function workspaceRun(workspaceId: string, executionId: string) {
+    return {
+      metadata: {
+        workspaceId,
+        correlation: {
+          executionId,
+          requestId: `request-${executionId}`,
+          source: 'workflow' as const,
+          workflowId: `workflow-${executionId}`,
+        },
+      },
+    }
+  }
+
+  beforeEach(() => {
+    mockResolveTriggerRegion.mockResolvedValue('us-east-1')
+    mockList.mockReset()
+    mockCancel.mockReset()
+    mockTrigger.mockReset()
+  })
+
+  it('cancels every workflow run enqueued for the workspace and nothing else', async () => {
+    const runs = installFakeTrigger()
+    const queue = new TriggerDevJobQueue()
+    await queue.enqueue('workflow-execution', {}, workspaceRun('removed', 'a'))
+    await queue.enqueue('webhook-execution', {}, workspaceRun('removed', 'b'))
+    await queue.enqueue('schedule-execution', {}, workspaceRun('removed', 'c'))
+    await queue.enqueue('resume-execution', {}, workspaceRun('removed', 'd'))
+    await queue.enqueue(
+      'workflow-group-cell',
+      {},
+      {
+        ...workspaceRun('removed', 'e'),
+        tags: ['tableId:t', 'rowId:r', 'group:g'],
+      }
+    )
+    await queue.enqueue('workflow-execution', {}, workspaceRun('kept', 'f'))
+    await queue.enqueue('cleanup-logs', {}, { metadata: { workspaceId: 'removed' } })
+
+    await expect(queue.cancelByWorkspace('removed', { maxRuns: 100 })).resolves.toEqual({
+      cancelledJobs: 5,
+      complete: true,
+    })
+
+    expect(runs.filter((run) => run.status === 'CANCELED').map((run) => run.id)).toEqual([
+      'run-0',
+      'run-1',
+      'run-2',
+      'run-3',
+      'run-4',
+    ])
+  })
+
+  it('stops at its run cap and reports the sweep incomplete until a later call finishes it', async () => {
+    const runs = installFakeTrigger()
+    const queue = new TriggerDevJobQueue()
+    for (let index = 0; index < 25; index++) {
+      await queue.enqueue('workflow-execution', {}, workspaceRun('removed', `run-${index}`))
+    }
+
+    await expect(queue.cancelByWorkspace('removed', { maxRuns: 10 })).resolves.toEqual({
+      cancelledJobs: 10,
+      complete: false,
+    })
+    await expect(queue.cancelByWorkspace('removed', { maxRuns: 10 })).resolves.toEqual({
+      cancelledJobs: 10,
+      complete: false,
+    })
+    await expect(queue.cancelByWorkspace('removed', { maxRuns: 10 })).resolves.toEqual({
+      cancelledJobs: 5,
+      complete: true,
+    })
+    expect(runs.every((run) => run.status === 'CANCELED')).toBe(true)
+  })
+
+  it('reports a sweep that reached exactly its cap as complete', async () => {
+    installFakeTrigger()
+    const queue = new TriggerDevJobQueue()
+    for (let index = 0; index < 10; index++) {
+      await queue.enqueue('workflow-execution', {}, workspaceRun('removed', `run-${index}`))
+    }
+
+    await expect(queue.cancelByWorkspace('removed', { maxRuns: 10 })).resolves.toEqual({
+      cancelledJobs: 10,
+      complete: true,
+    })
+  })
+
+  it('cancels the remaining runs before reporting one it could not cancel', async () => {
+    const runs = installFakeTrigger()
+    const queue = new TriggerDevJobQueue()
+    for (let index = 0; index < 12; index++) {
+      await queue.enqueue('workflow-execution', {}, workspaceRun('removed', `run-${index}`))
+    }
+    const cancel = mockCancel.getMockImplementation()
+    mockCancel.mockImplementation(async (runId: string) => {
+      if (runId === 'run-3') throw new MockApiError(500, { message: 'unavailable' })
+      return cancel?.(runId)
+    })
+
+    await expect(queue.cancelByWorkspace('removed', { maxRuns: 100 })).rejects.toThrow(
+      'unavailable'
+    )
+
+    expect(runs.filter((run) => run.status === 'CANCELED')).toHaveLength(11)
+  })
+
+  it('cancels at most ten runs at a time', async () => {
+    installFakeTrigger()
+    const queue = new TriggerDevJobQueue()
+    for (let index = 0; index < 35; index++) {
+      await queue.enqueue('workflow-execution', {}, workspaceRun('removed', `run-${index}`))
+    }
+    const cancel = mockCancel.getMockImplementation()
+    let active = 0
+    let maxActive = 0
+    mockCancel.mockImplementation(async (runId: string) => {
+      active += 1
+      maxActive = Math.max(maxActive, active)
+      await Promise.resolve()
+      active -= 1
+      return cancel?.(runId)
+    })
+
+    await queue.cancelByWorkspace('removed', { maxRuns: 100 })
+
+    expect(maxActive).toBe(10)
+  })
+
+  it('reports a failed listing so the caller retries', async () => {
+    mockList.mockImplementation(() => {
+      throw new MockApiError(503, { message: 'list unavailable' })
+    })
+    const queue = new TriggerDevJobQueue()
+
+    await expect(queue.cancelByWorkspace('removed', { maxRuns: 100 })).rejects.toThrow(
+      'list unavailable'
+    )
+  })
+})
+
+describe('TriggerDevJobQueue workspace cancellation past refusing runs', () => {
+  beforeEach(() => {
+    mockList.mockReset()
+    mockCancel.mockReset()
+  })
+
+  it('reaches later runs when the first ones keep refusing cancellation', async () => {
+    const listed = Array.from({ length: 15 }, (_, index) => ({
+      id: `run-${index}`,
+      tags: ['workspaceId:removed'],
+    }))
+    mockList.mockImplementation(() => createListPage(listed))
+    const refusing = new Set(['run-0', 'run-1', 'run-2'])
+    mockCancel.mockImplementation(async (runId: string) => {
+      if (refusing.has(runId)) throw new MockApiError(500, { message: 'refused' })
+    })
+    const queue = new TriggerDevJobQueue()
+
+    await expect(queue.cancelByWorkspace('removed', { maxRuns: 5 })).rejects.toThrow('refused')
+
+    expect(mockCancel).toHaveBeenCalledWith('run-9')
+  })
+})
+
+describe('TriggerDevJobQueue workspace cancellation of long-lived runs', () => {
+  beforeEach(() => {
+    mockList.mockReset()
+    mockCancel.mockReset().mockResolvedValue(undefined)
+  })
+
+  it('searches every active run of the workspace, however long ago it was created', async () => {
+    mockList.mockReturnValue(createListPage([{ id: 'run-old', tags: ['workspaceId:removed'] }]))
+    const queue = new TriggerDevJobQueue()
+
+    await queue.cancelByWorkspace('removed', { maxRuns: 100 })
+
+    const [filter] = mockList.mock.calls[0] as [Record<string, unknown>]
+    expect(filter.from).toBeUndefined()
+    expect(filter.period).toBeUndefined()
+    expect(filter.to).toBeInstanceOf(Date)
+    expect(mockCancel).toHaveBeenCalledWith('run-old')
+  })
+
+  it('falls back to the retention window when Trigger.dev refuses the unbounded search', async () => {
+    mockList
+      .mockImplementationOnce(() => {
+        throw new MockApiError(422, { message: 'narrow the time window' })
+      })
+      .mockReturnValueOnce(createListPage([{ id: 'run-recent', tags: ['workspaceId:removed'] }]))
+    const queue = new TriggerDevJobQueue()
+
+    await expect(queue.cancelByWorkspace('removed', { maxRuns: 100 })).resolves.toEqual({
+      cancelledJobs: 1,
+      complete: true,
+    })
+
+    const [fallback] = mockList.mock.calls[1] as [Record<string, unknown>]
+    expect(fallback.from).toBeInstanceOf(Date)
+    expect(mockCancel).toHaveBeenCalledWith('run-recent')
+  })
+})
diff --git a/apps/sim/lib/core/async-jobs/backends/trigger-dev.ts b/apps/sim/lib/core/async-jobs/backends/trigger-dev.ts
index 3ef4eaf5a61..a31ea5916b4 100644
--- a/apps/sim/lib/core/async-jobs/backends/trigger-dev.ts
+++ b/apps/sim/lib/core/async-jobs/backends/trigger-dev.ts
@@ -24,6 +24,9 @@ import {
   type JobType,
   TERMINAL_JOB_STATUSES,
   validateMaxDurationSeconds,
+  WORKSPACE_EXECUTION_JOB_TYPES,
+  type WorkspaceJobCancellationOptions,
+  type WorkspaceJobCancellationResult,
 } from '@/lib/core/async-jobs/types'
 import { recordExecutionCancellationBackendResult } from '@/lib/core/execution-limits/metrics'
 
@@ -38,6 +41,7 @@ const ACTIVE_TRIGGER_RUN_STATUSES = [
 ] as const
 const CANCELLATION_LIST_PAGE_SIZE = 25
 const CANCELLATION_OPERATION_CHUNK_SIZE = 10
+const WORKSPACE_CANCELLATION_LIST_PAGE_SIZE = 100
 function classifyTriggerEnqueueError(error: unknown): AsyncJobEnqueueError {
   if (error instanceof ApiError && error.status && error.status >= 400 && error.status < 500) {
     return new AsyncJobEnqueueError(toError(error).message, {
@@ -124,6 +128,10 @@ function buildExecutionTag(executionId: string): string {
   return tag.length <= 128 ? tag : `executionIdHash:${sha256Hex(executionId)}`
 }
 
+function buildWorkspaceTag(workspaceId: string): string {
+  return `workspaceId:${workspaceId}`
+}
+
 function buildWorkflowTag(workflowId: string): string {
   const tag = `workflowId:${workflowId}`
   return tag.length <= 128 ? tag : `workflowIdHash:${sha256Hex(workflowId)}`
@@ -168,10 +176,13 @@ interface CancellationListRun {
 }
 
 interface CancellationScanOptions {
-  binding: ExecutionJobBinding
+  /** Required by any candidate that sets `verifyPayload`. */
+  binding?: ExecutionJobBinding
   cancelJob: (jobId: string) => Promise<void>
   listRuns: () => AsyncIterable<CancellationListRun>
   selectCandidate: (run: CancellationListRun) => CancellationCandidate | undefined
+  /** Checked before each listed run; true ends the scan there. */
+  shouldStop?: () => boolean
   state: CancellationScanState
 }
 
@@ -182,7 +193,7 @@ function recordCancellationCandidateFailure(state: CancellationScanState, error:
 
 async function processCancellationCandidateBatch(
   candidates: readonly CancellationCandidate[],
-  binding: ExecutionJobBinding,
+  binding: ExecutionJobBinding | undefined,
   state: CancellationScanState,
   cancelJob: (jobId: string) => Promise<void>
 ): Promise<void> {
@@ -196,7 +207,7 @@ async function processCancellationCandidateBatch(
     uniqueCandidates.map(async (candidate) => {
       if (candidate.verifyPayload) {
         const legacyRun = await runs.retrieve(candidate.id)
-        if (!payloadMatchesExecution(legacyRun.payload, binding)) return false
+        if (!binding || !payloadMatchesExecution(legacyRun.payload, binding)) return false
       }
       await cancelJob(candidate.id)
       return true
@@ -228,6 +239,7 @@ async function scanAndCancelTriggerRuns(options: CancellationScanOptions): Promi
 
   try {
     for await (const run of options.listRuns()) {
+      if (options.shouldStop?.()) break
       const candidate = options.selectCandidate(run)
       if (!candidate) continue
       candidates.push(candidate)
@@ -240,6 +252,45 @@ async function scanAndCancelTriggerRuns(options: CancellationScanOptions): Promi
   }
 }
 
+/**
+ * Lists a workspace's active workflow runs however old they are: a removal must reach a run that
+ * has been queued or waiting for longer than any recent window. Trigger.dev searches only the
+ * last 7 days unless a time bound is given, and passing only `to` removes the lower bound. When
+ * Trigger.dev refuses that search as too expensive (422) before returning a run, the listing
+ * falls back to the queue-wait retention window rather than sweeping nothing.
+ */
+async function* listActiveWorkspaceRuns(
+  workspaceId: string,
+  workspaceTag: string
+): AsyncIterable<CancellationListRun> {
+  const filter = {
+    tag: workspaceTag,
+    taskIdentifier: [...WORKSPACE_EXECUTION_JOB_TYPES],
+    status: [...ACTIVE_TRIGGER_RUN_STATUSES],
+    limit: WORKSPACE_CANCELLATION_LIST_PAGE_SIZE,
+  }
+  let listed = false
+  try {
+    for await (const run of runs.list({ ...filter, to: new Date() })) {
+      listed = true
+      yield run
+    }
+    return
+  } catch (error) {
+    if (listed || !(error instanceof ApiError) || error.status !== 422) throw error
+    logger.warn(
+      'Trigger.dev refused an unbounded workspace run search; using the retention window',
+      {
+        workspaceId,
+      }
+    )
+  }
+  yield* runs.list({
+    ...filter,
+    from: new Date(Date.now() - JOB_PENDING_RETENTION_HOURS * 60 * 60 * 1000),
+  })
+}
+
 /**
  * Maps trigger.dev task IDs to our JobType
  */
@@ -657,6 +708,55 @@ export class TriggerDevJobQueue implements JobQueueBackend {
     }
   }
 
+  async cancelByWorkspace(
+    workspaceId: string,
+    options: WorkspaceJobCancellationOptions
+  ): Promise<WorkspaceJobCancellationResult> {
+    const workspaceTag = buildWorkspaceTag(workspaceId)
+    const state: CancellationScanState = { cancelledJobs: 0, failureCount: 0 }
+    let complete = true
+
+    await scanAndCancelTriggerRuns({
+      cancelJob: (jobId) => this.cancelJob(jobId),
+      listRuns: () => listActiveWorkspaceRuns(workspaceId, workspaceTag),
+      /**
+       * Only cancelled runs count toward the cap, so runs that keep refusing cancellation cannot
+       * hold every retry at the front of the listing; a separate bound on failures keeps one call
+       * finite. Both are counted per chunk, so a call may overshoot `maxRuns` by under a chunk.
+       */
+      shouldStop: () => {
+        if (
+          state.cancelledJobs < options.maxRuns &&
+          state.failureCount < options.maxRuns &&
+          !options.signal?.aborted
+        ) {
+          return false
+        }
+        complete = false
+        return true
+      },
+      selectCandidate: (run) =>
+        run.tags.includes(workspaceTag) ? { id: run.id, verifyPayload: false } : undefined,
+      state,
+    })
+
+    if (state.firstError) {
+      logger.error('Failed to cancel trigger.dev runs for workspace', {
+        workspaceId,
+        cancelledJobs: state.cancelledJobs,
+        failureCount: state.failureCount,
+        error: state.firstError.message,
+      })
+      throw state.firstError
+    }
+    logger.info('Cancelled trigger.dev runs for workspace', {
+      workspaceId,
+      cancelledJobs: state.cancelledJobs,
+      complete,
+    })
+    return { cancelledJobs: state.cancelledJobs, complete }
+  }
+
   cancelByKey(_cancelKey: string): boolean {
     // No in-process AbortControllers to abort — trigger.dev runs are cancelled
     // by jobId or via tag sweep (see `cancelCellRunsByTags`). Callers that
@@ -684,7 +784,7 @@ function buildTags(options?: EnqueueOptions): string[] {
   const workflowId =
     meta?.correlation?.workflowId ??
     (typeof meta?.workflowId === 'string' ? meta.workflowId : undefined)
-  if (meta?.workspaceId) tags.push(`workspaceId:${meta.workspaceId}`)
+  if (meta?.workspaceId) tags.push(buildWorkspaceTag(meta.workspaceId))
   if (workflowId) tags.push(buildWorkflowTag(workflowId))
   if (meta?.userId) tags.push(`userId:${meta.userId}`)
 
diff --git a/apps/sim/lib/core/async-jobs/types.ts b/apps/sim/lib/core/async-jobs/types.ts
index c65b5df56fd..8efb0e9893a 100644
--- a/apps/sim/lib/core/async-jobs/types.ts
+++ b/apps/sim/lib/core/async-jobs/types.ts
@@ -176,6 +176,31 @@ export const EXECUTION_JOB_TYPES_BY_CANCELLATION_SCOPE = {
   resume: ['resume-execution'],
 } as const satisfies Record<ExecutionJobCancellationScope, readonly JobType[]>
 
+/**
+ * Every job type that runs a workflow on a workspace's behalf. Each one is enqueued with
+ * `metadata.workspaceId`, so a workspace-wide stop can find them without knowing their executions.
+ */
+export const WORKSPACE_EXECUTION_JOB_TYPES = [
+  'workflow-execution',
+  'schedule-execution',
+  'webhook-execution',
+  'resume-execution',
+  'workflow-group-cell',
+] as const satisfies readonly JobType[]
+
+export interface WorkspaceJobCancellationOptions {
+  /** Most runs one call may inspect; the rest are left for the caller's next call. */
+  maxRuns: number
+  /** Stops the sweep between pages once the caller's time is up. */
+  signal?: AbortSignal
+}
+
+export interface WorkspaceJobCancellationResult {
+  cancelledJobs: number
+  /** False when the sweep stopped at `maxRuns` or its signal, so active runs may remain. */
+  complete: boolean
+}
+
 export type AsyncJobEnqueueAcceptance = 'rejected' | 'unknown'
 
 interface AsyncJobEnqueueErrorOptions {
@@ -311,6 +336,16 @@ export interface JobQueueBackend {
    * Returns `true` if a matching controller was found and aborted.
    */
   cancelByKey(cancelKey: string): boolean
+
+  /**
+   * Cancel every queued, delayed or running workflow job (`WORKSPACE_EXECUTION_JOB_TYPES`)
+   * enqueued for a workspace, for when the workspace itself is being removed. Throws when any
+   * listing or cancellation failed, after attempting the rest, so the caller can retry.
+   */
+  cancelByWorkspace(
+    workspaceId: string,
+    options: WorkspaceJobCancellationOptions
+  ): Promise<WorkspaceJobCancellationResult>
 }
 
 export type AsyncBackendType = 'trigger-dev' | 'database'
diff --git a/apps/sim/lib/core/outbox/handlers.ts b/apps/sim/lib/core/outbox/handlers.ts
index 77c088f6b52..240f019b531 100644
--- a/apps/sim/lib/core/outbox/handlers.ts
+++ b/apps/sim/lib/core/outbox/handlers.ts
@@ -14,6 +14,7 @@ import {
 import { MEMBER_BILLING_RECONCILIATION_EVENT_TYPE } from '@/lib/billing/organizations/membership-reconciliation-event'
 import { OUTBOX_EVENT_TYPES } from '@/lib/billing/webhooks/outbox-events'
 import type { LazyOutboxHandlerGroup } from '@/lib/core/outbox/service'
+import { EXECUTION_REMOVAL_STOP_EVENT } from '@/lib/execution/removal-stop-event'
 import { DIRECT_GRANT_EMAIL_EVENT_TYPE } from '@/lib/invitations/direct-grant-event'
 import { SLACK_SEARCH_TURN_EVENT } from '@/lib/knowledge/application/slack-search/turn-event'
 import {
@@ -186,6 +187,13 @@ export const OUTBOX_HANDLER_GROUPS: readonly LazyOutboxHandlerGroup[] = [
       return organizationResourceCleanupOutboxHandlers
     },
   },
+  {
+    events: [EXECUTION_REMOVAL_STOP_EVENT],
+    load: async () => {
+      const { executionRemovalStopOutboxHandlers } = await import('@/lib/execution/removal-stops')
+      return executionRemovalStopOutboxHandlers
+    },
+  },
   {
     events: [INBOX_CLEANUP_EVENT],
     load: async () => {
diff --git a/apps/sim/lib/execution/removal-stop-event.ts b/apps/sim/lib/execution/removal-stop-event.ts
new file mode 100644
index 00000000000..c2b01a49ccb
--- /dev/null
+++ b/apps/sim/lib/execution/removal-stop-event.ts
@@ -0,0 +1,2 @@
+/** Outbox event that stops the workflow runs of a removed workspace, workflow set or account. */
+export const EXECUTION_REMOVAL_STOP_EVENT = 'execution.removal-stop'
diff --git a/apps/sim/lib/execution/removal-stop-records.test.ts b/apps/sim/lib/execution/removal-stop-records.test.ts
new file mode 100644
index 00000000000..b3f43dcd2ec
--- /dev/null
+++ b/apps/sim/lib/execution/removal-stop-records.test.ts
@@ -0,0 +1,53 @@
+/**
+ * Failure modes:
+ * - the signal deadline passes before the Redis writes start, and the local runs are left going
+ * - a ban stops a previously archived workspace whose ownership moved on, and its new owner's
+ *   runs are stopped
+ */
+import { db } from '@sim/db'
+import { workspace } from '@sim/db/schema'
+import { dbChainMockFns, queueTableRows, resetDbChainMock } from '@sim/testing/mocks/database.mock'
+import { describe, expect, it, vi } from 'vitest'
+import {
+  registerManualExecutionAborter,
+  unregisterManualExecutionAborter,
+} from '@/lib/execution/manual-cancellation'
+import { recordExecutionStops, signalExecutionStops } from '@/lib/execution/removal-stop-records'
+
+describe('signalling removed executions', () => {
+  it('aborts the runs this process holds even after the signal deadline has passed', async () => {
+    const abort = vi.fn()
+    registerManualExecutionAborter('execution-local', abort)
+    const deadline = new AbortController()
+    deadline.abort()
+
+    try {
+      await expect(
+        signalExecutionStops(
+          [{ executionId: 'execution-local', executionDeadlineAt: null }],
+          deadline.signal
+        )
+      ).resolves.toBe(1)
+    } finally {
+      unregisterManualExecutionAborter('execution-local', abort)
+    }
+
+    expect(abort).toHaveBeenCalledOnce()
+  })
+})
+
+describe('recording a banned owner’s stops', () => {
+  it('records nothing for a workspace the user no longer owns', async () => {
+    resetDbChainMock()
+    queueTableRows(workspace, [])
+
+    await expect(
+      db.transaction((tx) =>
+        recordExecutionStops(tx, { workspaceIds: ['transferred'], ownerId: 'banned-user' })
+      )
+    ).resolves.toEqual([])
+
+    expect(dbChainMockFns.for).toHaveBeenCalledWith('share')
+    expect(dbChainMockFns.insert).not.toHaveBeenCalled()
+  })
+})
diff --git a/apps/sim/lib/execution/removal-stop-records.ts b/apps/sim/lib/execution/removal-stop-records.ts
new file mode 100644
index 00000000000..15452986fb6
--- /dev/null
+++ b/apps/sim/lib/execution/removal-stop-records.ts
@@ -0,0 +1,256 @@
+import type { db } from '@sim/db'
+import { resumeQueue, workflow, workflowExecutionLogs, workspace } from '@sim/db/schema'
+import { createLogger } from '@sim/logger'
+import { describeError } from '@sim/utils/errors'
+import { chunkArray } from '@sim/utils/helpers'
+import { and, asc, eq, inArray, or, type SQL } from 'drizzle-orm'
+import { enqueueOutboxEvents, MAX_BULK_ENQUEUE_EVENTS } from '@/lib/core/outbox/service'
+import { mapWithConcurrency } from '@/lib/core/utils/concurrency'
+import type { DbOrTx, DbTransaction } from '@/lib/db/types'
+import { markExecutionCancelled } from '@/lib/execution/cancellation'
+import { abortManualExecution } from '@/lib/execution/manual-cancellation'
+import { EXECUTION_REMOVAL_STOP_EVENT } from '@/lib/execution/removal-stop-event'
+
+const logger = createLogger('ExecutionRemovalStopRecords')
+
+/** Executions one event signals; bounds the payload of a workspace with many live runs. */
+const EXECUTIONS_PER_EVENT = 500
+const SIGNAL_CONCURRENCY = 10
+/** Bounds an inline signal pass; whatever it leaves, the outbox stop signals again. */
+const SIGNAL_DEADLINE_MS = 10_000
+
+/**
+ * What is being removed. Runs of a removed workspace or workflow lose the rows the ordinary
+ * cancel path reads (or, for an archive, the access they ran under), so they are stopped here:
+ * every in-flight execution gets the cancellation signal its engine watches, and every queued,
+ * delayed or running job enqueued for a removed workspace is cancelled in the job queue.
+ */
+interface ExecutionStopScope {
+  workspaceIds: readonly string[]
+  /** An account whose workflows are removed wherever they live, not only in `workspaceIds`. */
+  workflowsCreatedBy?: string
+}
+
+interface PlannedExecutionStop {
+  executionId: string
+  executionDeadlineAt: Date | null
+  workspaceId: string | null
+}
+
+/** Read before a removal erases the rows it names; enqueue with {@link enqueueExecutionStops}. */
+export interface ExecutionStopPlan {
+  workspaceIds: string[]
+  executions: PlannedExecutionStop[]
+}
+
+interface ExecutionStopPayload {
+  /** Workspace whose queued and running jobs this event sweeps. */
+  workspaceId?: string
+  executions: { executionId: string; executionDeadlineAt: string | null }[]
+}
+
+function scopeCondition(executor: DbOrTx, scope: ExecutionStopScope): SQL | undefined {
+  const conditions: SQL[] = []
+  if (scope.workspaceIds.length > 0) {
+    conditions.push(inArray(workflowExecutionLogs.workspaceId, [...scope.workspaceIds]))
+  }
+  if (scope.workflowsCreatedBy) {
+    conditions.push(
+      inArray(
+        workflowExecutionLogs.workflowId,
+        executor
+          .select({ id: workflow.id })
+          .from(workflow)
+          .where(eq(workflow.userId, scope.workflowsCreatedBy))
+      )
+    )
+  }
+  return conditions.length > 1 ? or(...conditions) : conditions[0]
+}
+
+/**
+ * Reads the executions a removal must stop. Run it in the removal's transaction (or, where the
+ * removal is not transactional, immediately before it): a cascade erases these rows. An active
+ * resume runs under its own execution id, so the claimed resume of a paused run is included.
+ */
+export async function collectExecutionStops(
+  executor: DbOrTx,
+  scope: ExecutionStopScope
+): Promise<ExecutionStopPlan> {
+  const workspaceIds = [...new Set(scope.workspaceIds)]
+  const inScope = scopeCondition(executor, scope)
+  if (!inScope) return { workspaceIds, executions: [] }
+
+  const running = await executor
+    .select({
+      executionId: workflowExecutionLogs.executionId,
+      executionDeadlineAt: workflowExecutionLogs.executionDeadlineAt,
+      workspaceId: workflowExecutionLogs.workspaceId,
+    })
+    .from(workflowExecutionLogs)
+    .where(and(eq(workflowExecutionLogs.status, 'running'), inScope))
+  const resuming = await executor
+    .select({
+      executionId: resumeQueue.newExecutionId,
+      executionDeadlineAt: workflowExecutionLogs.executionDeadlineAt,
+      workspaceId: workflowExecutionLogs.workspaceId,
+    })
+    .from(workflowExecutionLogs)
+    .innerJoin(
+      resumeQueue,
+      and(
+        eq(resumeQueue.parentExecutionId, workflowExecutionLogs.executionId),
+        eq(resumeQueue.status, 'claimed')
+      )
+    )
+    .where(and(inArray(workflowExecutionLogs.status, ['running', 'pending', 'paused']), inScope))
+
+  const executions = new Map<string, PlannedExecutionStop>()
+  for (const row of [...running, ...resuming]) {
+    if (!executions.has(row.executionId)) executions.set(row.executionId, row)
+  }
+  return { workspaceIds, executions: [...executions.values()] }
+}
+
+/**
+ * For a bare `DELETE FROM "user"`, which cascades every workspace the account owns (archived or
+ * not) and every workflow it created, wherever that workflow lives.
+ */
+export async function collectAccountRemovalStops(
+  executor: DbOrTx,
+  userId: string
+): Promise<ExecutionStopPlan> {
+  const owned = await executor
+    .select({ id: workspace.id })
+    .from(workspace)
+    .where(eq(workspace.ownerId, userId))
+  return collectExecutionStops(executor, {
+    workspaceIds: owned.map((row) => row.id),
+    workflowsCreatedBy: userId,
+  })
+}
+
+function toPayloadExecutions(
+  executions: readonly PlannedExecutionStop[]
+): ExecutionStopPayload['executions'] {
+  return executions.map(({ executionId, executionDeadlineAt }) => ({
+    executionId,
+    executionDeadlineAt: executionDeadlineAt?.toISOString() ?? null,
+  }))
+}
+
+/** One event per removed workspace carries its sweep; executions beyond one event's share spill over. */
+function buildPayloads(plan: ExecutionStopPlan): ExecutionStopPayload[] {
+  const byWorkspace = new Map<string, PlannedExecutionStop[]>(
+    plan.workspaceIds.map((workspaceId) => [workspaceId, []])
+  )
+  const elsewhere: PlannedExecutionStop[] = []
+  for (const execution of plan.executions) {
+    const removed = execution.workspaceId ? byWorkspace.get(execution.workspaceId) : undefined
+    if (removed) removed.push(execution)
+    else elsewhere.push(execution)
+  }
+
+  const payloads: ExecutionStopPayload[] = []
+  for (const [workspaceId, executions] of byWorkspace) {
+    const [first = [], ...rest] = chunkArray(executions, EXECUTIONS_PER_EVENT)
+    payloads.push({ workspaceId, executions: toPayloadExecutions(first) })
+    for (const chunk of rest) payloads.push({ executions: toPayloadExecutions(chunk) })
+  }
+  for (const chunk of chunkArray(elsewhere, EXECUTIONS_PER_EVENT)) {
+    payloads.push({ executions: toPayloadExecutions(chunk) })
+  }
+  return payloads
+}
+
+/**
+ * Records the stops as outbox events. In the removal's transaction they commit or roll back with
+ * it; the outbox worker retries each one until it is delivered.
+ */
+export async function enqueueExecutionStops(
+  executor: Pick<typeof db, 'insert'>,
+  plan: ExecutionStopPlan
+): Promise<void> {
+  for (const chunk of chunkArray(buildPayloads(plan), MAX_BULK_ENQUEUE_EVENTS)) {
+    await enqueueOutboxEvents(executor, EXECUTION_REMOVAL_STOP_EVENT, chunk)
+  }
+}
+
+/**
+ * {@link collectExecutionStops} then {@link enqueueExecutionStops}, in the caller's transaction.
+ * Returns the executions, to signal at once with {@link signalExecutionStops} after commit.
+ */
+export async function recordExecutionStops(
+  executor: DbTransaction,
+  scope: ExecutionStopScope & {
+    /**
+     * Stop only the workspaces this user still owns. Their rows are held `FOR SHARE` until the
+     * caller commits, so an ownership transfer racing the stop waits for it instead of having its
+     * new owner's runs stopped. Only workspace rows are locked, so no lock order is inverted.
+     */
+    ownerId?: string
+  }
+): Promise<ExecutionStopPlan['executions']> {
+  let { workspaceIds } = scope
+  if (scope.ownerId && workspaceIds.length > 0) {
+    const stillOwned = await executor
+      .select({ id: workspace.id })
+      .from(workspace)
+      .where(and(inArray(workspace.id, [...workspaceIds]), eq(workspace.ownerId, scope.ownerId)))
+      .orderBy(asc(workspace.id))
+      .for('share')
+    workspaceIds = stillOwned.map((row) => row.id)
+    if (workspaceIds.length === 0 && !scope.workflowsCreatedBy) return []
+  }
+  const plan = await collectExecutionStops(executor, { ...scope, workspaceIds })
+  await enqueueExecutionStops(executor, plan)
+  return plan.executions
+}
+
+/**
+ * Sets the cancellation flag each execution's engine watches, and aborts the ones this process
+ * runs, without the queue sweep or the retry the outbox event adds. Callers whose module graph
+ * cannot load the delivery stop in-flight runs with it at once and leave the rest to the outbox
+ * worker. The deadline (`signal`, else {@link SIGNAL_DEADLINE_MS}) bounds only the Redis writes;
+ * every execution this process runs is aborted first. Returns how many flags Redis refused or
+ * the deadline left unsent; never throws.
+ *
+ * Without Redis at all the flag is process-local by design (`markExecutionCancelled`) and a
+ * retry from any process could reach no further, so that outcome does not count as refused.
+ */
+export async function signalExecutionStops(
+  executions: readonly Pick<PlannedExecutionStop, 'executionId' | 'executionDeadlineAt'>[],
+  signal?: AbortSignal
+): Promise<number> {
+  for (const { executionId } of executions) {
+    try {
+      abortManualExecution(executionId)
+    } catch (error) {
+      logger.warn('Failed to abort a local execution being stopped', {
+        executionId,
+        error: describeError(error),
+      })
+    }
+  }
+  const deadline = signal ?? AbortSignal.timeout(SIGNAL_DEADLINE_MS)
+  let refused = 0
+  await mapWithConcurrency(executions, SIGNAL_CONCURRENCY, async (execution) => {
+    if (deadline.aborted) {
+      refused += 1
+      return
+    }
+    try {
+      const result = await markExecutionCancelled(execution.executionId, {
+        executionDeadlineAt: execution.executionDeadlineAt,
+      })
+      if (!result.durablyRecorded && result.reason !== 'redis_unavailable') refused += 1
+    } catch (error) {
+      refused += 1
+      logger.warn('Failed to signal an execution being stopped', {
+        executionId: execution.executionId,
+        error: describeError(error),
+      })
+    }
+  })
+  return refused
+}
diff --git a/apps/sim/lib/execution/removal-stops.integration.ts b/apps/sim/lib/execution/removal-stops.integration.ts
new file mode 100644
index 00000000000..a74d38b518e
--- /dev/null
+++ b/apps/sim/lib/execution/removal-stops.integration.ts
@@ -0,0 +1,369 @@
+/**
+ * Removing an account, banning it, or deleting its user row directly stops the workflow runs the
+ * removal orphans, against real PostgreSQL and Redis through the production removal paths (for
+ * `/admin/remove-user`, the handlers of its Better Auth delete hooks around a bare delete). A
+ * stopped execution is one whose cancellation flag, the one its engine watches, is set in Redis;
+ * a stopped queued job is a cancelled `async_jobs` row (the database queue these suites run on).
+ */
+import { authMock } from '@sim/testing/mocks/auth.mock'
+import { afterAll, describe, expect, it, vi } from 'vitest'
+
+const { redisUrl, inheritedRedisUrl } = await vi.hoisted(async () => {
+  const { readTestRedisUrl } = await import('@sim/db/testing/test-infrastructure')
+  const url = readTestRedisUrl()
+  const inheritedRedisUrl = process.env.REDIS_URL
+  if (url) process.env.REDIS_URL = url
+  return { redisUrl: url, inheritedRedisUrl }
+})
+
+vi.mock('@/lib/auth', () => authMock)
+
+import { db } from '@sim/db'
+import {
+  asyncJobs,
+  outboxEvent,
+  pausedExecutions,
+  permissions,
+  resumeQueue,
+  user,
+  workflow,
+  workflowExecutionLogs,
+  workflowExecutionSnapshots,
+  workspace,
+} from '@sim/db/schema'
+import { deleteWorkspaceFixture, insertWorkspaceFixture } from '@sim/db/testing/workspace-fixtures'
+import { generateId } from '@sim/utils/id'
+import { and, eq, inArray, sql } from 'drizzle-orm'
+import { holdAdminRemovalStops, stopAdminRemovedRuns } from '@/lib/auth/admin-removal-stops'
+import { closeRedisConnection, getRedisClient } from '@/lib/core/config/redis'
+import { processOutboxEventById } from '@/lib/core/outbox/service'
+import { EXECUTION_REMOVAL_STOP_EVENT } from '@/lib/execution/removal-stop-event'
+import { executionRemovalStopOutboxHandlers } from '@/lib/execution/removal-stops'
+import { deleteUserAccount } from '@/lib/users/account-deletion'
+import { disableUserResources } from '@/lib/workflows/lifecycle'
+
+const userIds: string[] = []
+const workspaceIds: string[] = []
+const snapshotIds: string[] = []
+const executionIds: string[] = []
+const jobIds: string[] = []
+
+async function seedUser() {
+  const id = generateId()
+  userIds.push(id)
+  const now = new Date()
+  await db.insert(user).values({
+    id,
+    name: 'Execution removal fixture',
+    email: `${id}@execution-removal.test`,
+    emailVerified: true,
+    createdAt: now,
+    updatedAt: now,
+  })
+  return id
+}
+
+async function seedWorkspace(ownerId: string, writers: string[] = []) {
+  const id = generateId()
+  workspaceIds.push(id)
+  await insertWorkspaceFixture(db, {
+    id,
+    name: 'Execution removal fixture',
+    ownerId,
+    billedAccountUserId: ownerId,
+  })
+  await db.insert(permissions).values(
+    [ownerId, ...writers].map((userId) => ({
+      id: generateId(),
+      userId,
+      entityType: 'workspace',
+      entityId: id,
+      permissionType: userId === ownerId ? ('admin' as const) : ('write' as const),
+    }))
+  )
+  return id
+}
+
+async function seedWorkflow(userId: string, workspaceId: string) {
+  const id = generateId()
+  const now = new Date()
+  await db.insert(workflow).values({
+    id,
+    userId,
+    workspaceId,
+    name: `Execution removal fixture ${id}`,
+    lastSynced: now,
+    createdAt: now,
+    updatedAt: now,
+  })
+  return id
+}
+
+/** A run whose engine is still executing: its log row says `running`. */
+async function seedExecution(
+  workspaceId: string,
+  workflowId: string,
+  status: 'running' | 'pending' | 'paused' = 'running'
+) {
+  const snapshotId = generateId()
+  snapshotIds.push(snapshotId)
+  await db.insert(workflowExecutionSnapshots).values({
+    id: snapshotId,
+    workflowId,
+    stateHash: snapshotId,
+    stateData: {},
+  })
+  const executionId = generateId()
+  executionIds.push(executionId)
+  await db.insert(workflowExecutionLogs).values({
+    id: generateId(),
+    workflowId,
+    workspaceId,
+    executionId,
+    stateSnapshotId: snapshotId,
+    level: 'info',
+    status,
+    trigger: 'manual',
+    startedAt: new Date(),
+    executionData: {},
+  })
+  return executionId
+}
+
+/** A paused run whose resume a worker has claimed and is executing under its own id. */
+async function seedActiveResume(
+  workspaceId: string,
+  workflowId: string,
+  parentStatus: 'pending' | 'paused' = 'pending'
+) {
+  const parentExecutionId = await seedExecution(workspaceId, workflowId, parentStatus)
+  const pausedExecutionId = generateId()
+  await db.insert(pausedExecutions).values({
+    id: pausedExecutionId,
+    workflowId,
+    executionId: parentExecutionId,
+    executionSnapshot: {},
+    pausePoints: {},
+    totalPauseCount: 1,
+    status: 'paused',
+  })
+  const resumeExecutionId = generateId()
+  executionIds.push(resumeExecutionId)
+  await db.insert(resumeQueue).values({
+    id: generateId(),
+    pausedExecutionId,
+    parentExecutionId,
+    newExecutionId: resumeExecutionId,
+    contextId: generateId(),
+    status: 'claimed',
+  })
+  return resumeExecutionId
+}
+
+/** A job the queue has not started yet. */
+async function seedQueuedJob(workspaceId: string, type = 'webhook-execution') {
+  const id = `run_${generateId()}`
+  jobIds.push(id)
+  await db.insert(asyncJobs).values({
+    id,
+    type,
+    payload: {},
+    status: 'pending',
+    metadata: { workspaceId },
+  })
+  return id
+}
+
+async function isCancelled(executionId: string) {
+  return (await getRedisClient()?.exists(`execution:cancel:${executionId}`)) === 1
+}
+
+async function jobStatus(jobId: string) {
+  const [row] = await db
+    .select({ status: asyncJobs.status })
+    .from(asyncJobs)
+    .where(eq(asyncJobs.id, jobId))
+  return row?.status
+}
+
+/** Every stop event naming one of this test's executions or workspaces. */
+async function stopEvents(ids: string[]) {
+  const rows = await db
+    .select({ id: outboxEvent.id, status: outboxEvent.status, payload: outboxEvent.payload })
+    .from(outboxEvent)
+    .where(eq(outboxEvent.eventType, EXECUTION_REMOVAL_STOP_EVENT))
+  return rows.filter((row) => ids.some((id) => JSON.stringify(row.payload).includes(id)))
+}
+
+/** Processes the stop events the way the outbox worker does. */
+async function runOutboxWorker(eventIds: string[]) {
+  for (const eventId of eventIds) {
+    await processOutboxEventById(eventId, executionRemovalStopOutboxHandlers)
+  }
+}
+
+afterAll(async () => {
+  if (inheritedRedisUrl === undefined) Reflect.deleteProperty(process.env, 'REDIS_URL')
+  else process.env.REDIS_URL = inheritedRedisUrl
+})
+
+describe.runIf(Boolean(redisUrl))('removals stop the workflow runs they orphan', () => {
+  afterAll(async () => {
+    const redis = getRedisClient()
+    if (redis && executionIds.length > 0) {
+      await redis.del(...executionIds.map((id) => `execution:cancel:${id}`))
+    }
+    const scoped = [...executionIds, ...workspaceIds]
+    for (const id of scoped) {
+      await db
+        .delete(outboxEvent)
+        .where(
+          and(
+            eq(outboxEvent.eventType, EXECUTION_REMOVAL_STOP_EVENT),
+            sql`${outboxEvent.payload}::text like ${`%${id}%`}`
+          )
+        )
+    }
+    if (jobIds.length) await db.delete(asyncJobs).where(inArray(asyncJobs.id, jobIds))
+    if (executionIds.length) {
+      await db
+        .delete(workflowExecutionLogs)
+        .where(inArray(workflowExecutionLogs.executionId, executionIds))
+    }
+    if (snapshotIds.length) {
+      await db
+        .delete(workflowExecutionSnapshots)
+        .where(inArray(workflowExecutionSnapshots.id, snapshotIds))
+    }
+    if (workspaceIds.length) await deleteWorkspaceFixture(db, inArray(workspace.id, workspaceIds))
+    if (userIds.length) {
+      await db.delete(permissions).where(inArray(permissions.userId, userIds))
+      await db.delete(user).where(inArray(user.id, userIds))
+    }
+    await closeRedisConnection()
+  })
+
+  it('stops the running, resuming and queued runs of an erased account’s workspaces, and no one else’s', async () => {
+    const departingId = await seedUser()
+    const bystanderId = await seedUser()
+    const doomedWorkspaceId = await seedWorkspace(departingId)
+    const doomedWorkflowId = await seedWorkflow(departingId, doomedWorkspaceId)
+    const running = await seedExecution(doomedWorkspaceId, doomedWorkflowId)
+    const resuming = await seedActiveResume(doomedWorkspaceId, doomedWorkflowId)
+    const resumingLegacyPaused = await seedActiveResume(
+      doomedWorkspaceId,
+      await seedWorkflow(departingId, doomedWorkspaceId),
+      'paused'
+    )
+    const queued = await seedQueuedJob(doomedWorkspaceId)
+    const unrelatedJob = await seedQueuedJob(doomedWorkspaceId, 'cleanup-logs')
+    const keptWorkspaceId = await seedWorkspace(bystanderId)
+    const keptRunning = await seedExecution(
+      keptWorkspaceId,
+      await seedWorkflow(bystanderId, keptWorkspaceId)
+    )
+    const keptQueued = await seedQueuedJob(keptWorkspaceId)
+
+    await deleteUserAccount(departingId)
+
+    await expect.poll(() => isCancelled(running)).toBe(true)
+    await expect.poll(() => isCancelled(resuming)).toBe(true)
+    await expect.poll(() => isCancelled(resumingLegacyPaused)).toBe(true)
+    const recorded = await stopEvents([doomedWorkspaceId])
+    expect(recorded.map((event) => event.status)).toEqual(['pending'])
+
+    await runOutboxWorker(recorded.map((event) => event.id))
+
+    expect(await jobStatus(queued)).toBe('cancelled')
+    expect(await jobStatus(unrelatedJob)).toBe('pending')
+    expect((await stopEvents([doomedWorkspaceId])).map((event) => event.status)).toEqual([
+      'completed',
+    ])
+    expect(await isCancelled(keptRunning)).toBe(false)
+    expect(await jobStatus(keptQueued)).toBe('pending')
+  })
+
+  it('stops a banned owner’s running runs at once and leaves its queued runs to the outbox worker', async () => {
+    const bannedId = await seedUser()
+    const bystanderId = await seedUser()
+    const ownedWorkspaceId = await seedWorkspace(bannedId)
+    const running = await seedExecution(
+      ownedWorkspaceId,
+      await seedWorkflow(bannedId, ownedWorkspaceId)
+    )
+    const queued = await seedQueuedJob(ownedWorkspaceId)
+    const earlierArchivedId = await seedWorkspace(bannedId)
+    const runningInEarlierArchived = await seedExecution(
+      earlierArchivedId,
+      await seedWorkflow(bannedId, earlierArchivedId)
+    )
+    await db
+      .update(workspace)
+      .set({ archivedAt: new Date() })
+      .where(eq(workspace.id, earlierArchivedId))
+    const keptWorkspaceId = await seedWorkspace(bystanderId, [bannedId])
+    const keptRunning = await seedExecution(
+      keptWorkspaceId,
+      await seedWorkflow(bystanderId, keptWorkspaceId)
+    )
+    const keptQueued = await seedQueuedJob(keptWorkspaceId)
+
+    await disableUserResources(bannedId)
+
+    const [archived] = await db
+      .select({ archivedAt: workspace.archivedAt })
+      .from(workspace)
+      .where(eq(workspace.id, ownedWorkspaceId))
+    expect(archived.archivedAt).not.toBeNull()
+    expect(await isCancelled(running)).toBe(true)
+    expect(await isCancelled(runningInEarlierArchived)).toBe(true)
+    expect(await isCancelled(keptRunning)).toBe(false)
+    const recorded = await stopEvents([ownedWorkspaceId])
+    expect(recorded.map((event) => event.status)).toEqual(['pending'])
+    /** Admitted before the archive, logged after the stop was recorded. */
+    const loggedLate = await seedExecution(
+      ownedWorkspaceId,
+      await seedWorkflow(bannedId, ownedWorkspaceId)
+    )
+
+    await runOutboxWorker(recorded.map((event) => event.id))
+
+    expect(await isCancelled(loggedLate)).toBe(true)
+    expect(await jobStatus(queued)).toBe('cancelled')
+    expect(await jobStatus(keptQueued)).toBe('pending')
+  })
+
+  /**
+   * Drives the handlers Better Auth runs as the user delete's `before` and `after` hooks around
+   * the bare delete `/admin/remove-user` issues.
+   */
+  it('stops the runs of workflows a directly deleted user created in someone else’s workspace', async () => {
+    const removedId = await seedUser()
+    const bystanderId = await seedUser()
+    const sharedWorkspaceId = await seedWorkspace(bystanderId, [removedId])
+    const removedUsersRun = await seedExecution(
+      sharedWorkspaceId,
+      await seedWorkflow(removedId, sharedWorkspaceId)
+    )
+    const bystandersRun = await seedExecution(
+      sharedWorkspaceId,
+      await seedWorkflow(bystanderId, sharedWorkspaceId)
+    )
+    const bystandersQueued = await seedQueuedJob(sharedWorkspaceId)
+
+    const removed = { id: removedId }
+    await holdAdminRemovalStops(removed)
+    await db.delete(user).where(eq(user.id, removedId))
+    await stopAdminRemovedRuns(removed)
+
+    expect(await isCancelled(removedUsersRun)).toBe(true)
+    expect(await isCancelled(bystandersRun)).toBe(false)
+
+    await runOutboxWorker((await stopEvents([removedUsersRun])).map((event) => event.id))
+
+    expect(await jobStatus(bystandersQueued)).toBe('pending')
+    expect((await stopEvents([removedUsersRun])).map((event) => event.status)).toEqual([
+      'completed',
+    ])
+  })
+})
diff --git a/apps/sim/lib/execution/removal-stops.test.ts b/apps/sim/lib/execution/removal-stops.test.ts
new file mode 100644
index 00000000000..f33e7dff222
--- /dev/null
+++ b/apps/sim/lib/execution/removal-stops.test.ts
@@ -0,0 +1,97 @@
+/**
+ * The outbox outcome of a removal stop decides whether a run keeps going: an outcome that
+ * completes too early drops the stop, one that never completes dead-letters it. Failure modes:
+ * - Redis refuses the cancellation write → retry, and the queue sweep still runs
+ * - no Redis at all → the process-local flag is all there is, so the stop completes
+ * - a queue sweep fails → retry
+ * - one queue fails to load → the other is still swept, then retry
+ * - a sweep stops at its cap having cancelled runs → continue without spending an attempt
+ * - a sweep stops at its cap having cancelled nothing → retry, so it cannot loop forever
+ */
+import { asyncJobsMock, asyncJobsMockFns } from '@sim/testing/mocks/async-jobs.mock'
+import { redisConfigMockFns } from '@sim/testing/mocks/redis-config.mock'
+import { beforeEach, describe, expect, it, vi } from 'vitest'
+
+vi.mock('@/lib/core/async-jobs', () => asyncJobsMock)
+
+import type { OutboxEventContext } from '@/lib/core/outbox/service'
+import { EXECUTION_REMOVAL_STOP_EVENT } from '@/lib/execution/removal-stop-event'
+import { executionRemovalStopOutboxHandlers } from '@/lib/execution/removal-stops'
+
+const { mockJobQueue } = asyncJobsMockFns
+const stopRemovedExecutions = executionRemovalStopOutboxHandlers[EXECUTION_REMOVAL_STOP_EVENT]
+
+function context(): OutboxEventContext {
+  return {
+    eventId: 'event-1',
+    eventType: EXECUTION_REMOVAL_STOP_EVENT,
+    attempts: 0,
+    maxAttempts: 10,
+    signal: new AbortController().signal,
+    checkpointPayload: vi.fn(),
+  }
+}
+
+const payload = {
+  workspaceId: 'workspace-1',
+  executions: [{ executionId: 'execution-1', executionDeadlineAt: null }],
+}
+
+describe('execution removal stop outcome', () => {
+  beforeEach(() => {
+    redisConfigMockFns.mockGetRedisClient.mockReturnValue(null)
+    mockJobQueue.cancelByWorkspace.mockReset().mockResolvedValue({
+      cancelledJobs: 0,
+      complete: true,
+    })
+  })
+
+  it('completes when there is no Redis to record the cancellation in', async () => {
+    await expect(stopRemovedExecutions(payload, context())).resolves.toBeUndefined()
+  })
+
+  it('retries when Redis refuses the cancellation, after still sweeping the queue', async () => {
+    redisConfigMockFns.mockGetRedisClient.mockReturnValue({
+      eval: vi.fn().mockRejectedValue(new Error('READONLY')),
+      publish: vi.fn().mockResolvedValue(1),
+    })
+
+    await expect(stopRemovedExecutions(payload, context())).rejects.toThrow('not recorded')
+    expect(mockJobQueue.cancelByWorkspace).toHaveBeenCalledWith('workspace-1', expect.anything())
+  })
+
+  it('retries when the queue sweep fails', async () => {
+    mockJobQueue.cancelByWorkspace.mockRejectedValue(new Error('provider unavailable'))
+
+    await expect(stopRemovedExecutions(payload, context())).rejects.toThrow('provider unavailable')
+  })
+
+  it('still sweeps the database queue when the primary queue fails to load, then retries', async () => {
+    asyncJobsMockFns.mockGetJobQueue.mockRejectedValueOnce(new Error('queue unavailable'))
+
+    await expect(stopRemovedExecutions(payload, context())).rejects.toThrow('queue unavailable')
+    expect(mockJobQueue.cancelByWorkspace).toHaveBeenCalledWith('workspace-1', expect.anything())
+  })
+
+  it('continues without spending an attempt when the sweep stops at its cap after progress', async () => {
+    mockJobQueue.cancelByWorkspace.mockResolvedValue({ cancelledJobs: 1000, complete: false })
+
+    await expect(stopRemovedExecutions(payload, context())).resolves.toMatchObject({
+      outcome: 'deferred',
+      consumeAttempt: false,
+    })
+  })
+
+  it('retries when the sweep stops without cancelling anything', async () => {
+    mockJobQueue.cancelByWorkspace.mockResolvedValue({ cancelledJobs: 0, complete: false })
+
+    await expect(stopRemovedExecutions(payload, context())).rejects.toThrow('no progress')
+  })
+
+  it('sweeps no queue for executions outside a removed workspace', async () => {
+    await expect(
+      stopRemovedExecutions({ executions: payload.executions }, context())
+    ).resolves.toBeUndefined()
+    expect(mockJobQueue.cancelByWorkspace).not.toHaveBeenCalled()
+  })
+})
diff --git a/apps/sim/lib/execution/removal-stops.ts b/apps/sim/lib/execution/removal-stops.ts
new file mode 100644
index 00000000000..93e6de89801
--- /dev/null
+++ b/apps/sim/lib/execution/removal-stops.ts
@@ -0,0 +1,101 @@
+import { db } from '@sim/db'
+import { z } from 'zod'
+import { getInlineJobQueue, getJobQueue } from '@/lib/core/async-jobs'
+import type { JobQueueBackend } from '@/lib/core/async-jobs/types'
+import {
+  continueOutboxHandler,
+  type OutboxHandler,
+  type OutboxHandlerRegistry,
+  withOutboxHandlerTimeout,
+} from '@/lib/core/outbox/service'
+import { EXECUTION_REMOVAL_STOP_EVENT } from '@/lib/execution/removal-stop-event'
+import { collectExecutionStops, signalExecutionStops } from '@/lib/execution/removal-stop-records'
+
+/** Queue runs one handler attempt cancels before handing the rest to its next run. */
+const SWEEP_MAX_RUNS_PER_ATTEMPT = 1_000
+const HANDLER_TIMEOUT_MS = 5 * 60 * 1000
+
+const payloadSchema = z.object({
+  workspaceId: z.string().min(1).optional(),
+  executions: z.array(
+    z.object({
+      executionId: z.string().min(1),
+      executionDeadlineAt: z
+        .string()
+        .nullable()
+        .transform((deadline) => (deadline ? new Date(deadline) : null)),
+    })
+  ),
+})
+
+/**
+ * Sweeps the primary queue and the database queue that inline webhook runs use even when
+ * Trigger.dev is primary. Each queue that loads is swept, even when the other fails to load or
+ * to sweep, before the first failure is rethrown for the outbox to retry.
+ */
+async function sweepWorkspaceJobs(workspaceId: string, signal: AbortSignal) {
+  const resolved = await Promise.allSettled([getJobQueue(), getInlineJobQueue()])
+  const queues = new Set<JobQueueBackend>()
+  const failures: unknown[] = []
+  for (const result of resolved) {
+    if (result.status === 'fulfilled') queues.add(result.value)
+    else failures.push(result.reason)
+  }
+  const results = await Promise.allSettled(
+    [...queues].map((queue) =>
+      queue.cancelByWorkspace(workspaceId, { maxRuns: SWEEP_MAX_RUNS_PER_ATTEMPT, signal })
+    )
+  )
+  let cancelledJobs = 0
+  let complete = true
+  for (const result of results) {
+    if (result.status === 'rejected') failures.push(result.reason)
+    else {
+      cancelledJobs += result.value.cancelledJobs
+      complete &&= result.value.complete
+    }
+  }
+  if (failures.length > 0) throw failures[0]
+  return { cancelledJobs, complete }
+}
+
+/**
+ * Idempotent, so a retry repeats the whole event: a repeated signal and a repeated cancel of a
+ * run that already ended are both no-ops.
+ */
+const stopRemovedExecutions: OutboxHandler<unknown> = withOutboxHandlerTimeout(
+  async (rawPayload, context) => {
+    const payload = payloadSchema.parse(rawPayload)
+    const executions = new Map(
+      payload.executions.map((execution) => [execution.executionId, execution])
+    )
+    /**
+     * An archived workspace keeps its rows, so a run that passed admission before the archive
+     * but logged after the stop was recorded is found here. An erased workspace has none left.
+     */
+    if (payload.workspaceId) {
+      const { executions: live } = await collectExecutionStops(db, {
+        workspaceIds: [payload.workspaceId],
+      })
+      for (const execution of live) executions.set(execution.executionId, execution)
+    }
+    const [unrecorded, sweep] = await Promise.all([
+      signalExecutionStops([...executions.values()], context.signal),
+      payload.workspaceId
+        ? sweepWorkspaceJobs(payload.workspaceId, context.signal)
+        : { cancelledJobs: 0, complete: true },
+    ])
+    if (unrecorded > 0) {
+      throw new Error(`${unrecorded} execution cancellations were not recorded`)
+    }
+    if (!sweep.complete) {
+      if (sweep.cancelledJobs === 0) throw new Error('Workspace job sweep made no progress')
+      return continueOutboxHandler('Workspace job sweep reached its per-attempt cap')
+    }
+  },
+  HANDLER_TIMEOUT_MS
+)
+
+export const executionRemovalStopOutboxHandlers = {
+  [EXECUTION_REMOVAL_STOP_EVENT]: stopRemovedExecutions,
+} satisfies OutboxHandlerRegistry
diff --git a/apps/sim/lib/users/account-deletion.ts b/apps/sim/lib/users/account-deletion.ts
index ca0c883f95f..2cc251e0b1b 100644
--- a/apps/sim/lib/users/account-deletion.ts
+++ b/apps/sim/lib/users/account-deletion.ts
@@ -39,6 +39,11 @@ import {
 } from '@/lib/billing/organizations/resource-handoff'
 import { OrchestrationError } from '@/lib/core/orchestration/types'
 import type { DbOrTx } from '@/lib/db/types'
+import {
+  type ExecutionStopPlan,
+  recordExecutionStops,
+  signalExecutionStops,
+} from '@/lib/execution/removal-stop-records'
 import { type StoppedRun, stopLiveRunsTx } from '@/lib/mothership/async-runs/repository'
 import { enqueueMemoryCleanup } from '@/lib/mothership/memories/cleanup-outbox'
 import { deliverRunStops } from '@/lib/mothership/request/application/stop-delivery'
@@ -782,6 +787,7 @@ export async function deleteUserAccount(userId: string): Promise<AccountDeletion
   let cancelledDispatches: CancelledDispatch[] = []
   let cancelledMarkers: CancelledCellMarker[] = []
   let stoppedRuns: StoppedRun[] = []
+  let executionStops: ExecutionStopPlan['executions'] = []
   /**
    * Resolved while the account's settings still exist to route a superuser's runs. It only
    * routes the best-effort push, so a failed lookup falls back to the default endpoint.
@@ -862,6 +868,14 @@ export async function deleteUserAccount(userId: string): Promise<AccountDeletion
      */
     stoppedRuns = await stopLiveRunsTx(tx, { userId })
     if (doomedWorkspaceIds.length > 0) {
+      /**
+       * Workflow runs in the doomed workspaces lose their log rows to the cascade below, after
+       * which no cancel path can find them, and they would keep spending hosted keys for an
+       * account that no longer exists. The workspace rows are locked, so no new run can log
+       * here before commit; the stops commit with the erasure. After commit the in-flight runs
+       * are signalled at once and the outbox worker sweeps the job queue.
+       */
+      executionStops = await recordExecutionStops(tx, { workspaceIds: doomedWorkspaceIds })
       /**
        * Re-checked here rather than trusted from the plan: a workspace that
        * gained a member since the preview is no longer private, and deleting it
@@ -1013,6 +1027,13 @@ export async function deleteUserAccount(userId: string): Promise<AccountDeletion
     }
   })
 
+  /**
+   * Not awaited: every second a run keeps going spends hosted keys, and nothing below should
+   * wait on Redis. It never throws and stops at its own deadline; the committed outbox stops
+   * repeat it and sweep the job queue.
+   */
+  void signalExecutionStops(executionStops)
+
   await invalidateReportingUsagePrefixes()
 
   await announceCancelledTableWork(cancelledDispatches, cancelledMarkers)
diff --git a/apps/sim/lib/workflows/lifecycle.ts b/apps/sim/lib/workflows/lifecycle.ts
index 14fef2a268e..f5e2269e452 100644
--- a/apps/sim/lib/workflows/lifecycle.ts
+++ b/apps/sim/lib/workflows/lifecycle.ts
@@ -292,8 +292,16 @@ export async function archiveWorkflowsByIdsInWorkspace(
 
 /**
  * Disables all resources owned by a banned user by archiving every workspace
- * they own (cascading to workflows, chats, KBs, tables, files, etc.)
- * and deleting their personal API keys.
+ * they own (cascading to workflows, chats, KBs, tables, files, etc.),
+ * deleting their personal API keys, and stopping the workflow runs in progress
+ * in every workspace they own once it is archived, including ones archived
+ * earlier. Each archive records Stop for its Chat runs, which they read on their
+ * next run-control call.
+ *
+ * Runs the user started in workspaces they do not own keep going. The log row
+ * has no actor column, and the user it records is the billed account for
+ * schedule and webhook runs, so it cannot tell whose run it is; revoking the
+ * keys and sessions stops the user starting more.
  */
 export async function disableUserResources(userId: string): Promise<void> {
   const requestId = generateRequestId()
@@ -304,20 +312,60 @@ export async function disableUserResources(userId: string): Promise<void> {
   /** Revoked first: the keys are the ban's access boundary, and an archive may fail. */
   await db.delete(apiKey).where(eq(apiKey.userId, userId))
 
-  const ownedWorkspaces = await db
-    .select({ id: workspace.id })
+  const owned = await db
+    .select({ id: workspace.id, archivedAt: workspace.archivedAt })
     .from(workspace)
-    .where(and(eq(workspace.ownerId, userId), isNull(workspace.archivedAt)))
+    .where(eq(workspace.ownerId, userId))
+  const ownedWorkspaces = owned.filter((row) => !row.archivedAt)
+
+  const { recordExecutionStops, signalExecutionStops } = await import(
+    '@/lib/execution/removal-stop-records'
+  )
+  /**
+   * Records the workspaces' stops and signals their in-flight runs; the recorded stop's queue
+   * sweep and retries run in the outbox worker, since this module sits in page graphs that must
+   * not load that delivery. Logged, not thrown: the ban and archives have committed, and banning
+   * the user again records the stops again, since every owned workspace is stopped, archived or
+   * not.
+   */
+  let stopFailed = false
+  const stopRuns = async (workspaceIds: string[]): Promise<void> => {
+    if (workspaceIds.length === 0) return
+    try {
+      const executions = await db.transaction((tx) =>
+        recordExecutionStops(tx, { workspaceIds, ownerId: userId })
+      )
+      await signalExecutionStops(executions)
+    } catch (error) {
+      stopFailed = true
+      logger.error(`[${requestId}] Failed to stop runs for banned user ${userId}`, {
+        workspaceIds,
+        error: getErrorMessage(error),
+      })
+    }
+  }
 
   const failures: { workspaceId: string; error: unknown }[] = []
+  /** Workspaces archived before this ban still have the runs an archive leaves going. */
+  const stopping = [stopRuns(owned.filter((row) => row.archivedAt).map((row) => row.id))]
   // Sequential: environments of one Project contend on its exclusive lock.
   for (const row of ownedWorkspaces) {
     try {
-      await archiveWorkspace(row.id, { requestId, expectedOwnerId: userId })
+      await archiveWorkspace(row.id, {
+        requestId,
+        expectedOwnerId: userId,
+        /** Called once the archive commits, before its slower post-commit cleanup. */
+        deliverStops: () => {
+          stopping.push(stopRuns([row.id]))
+        },
+      })
     } catch (error) {
       failures.push({ workspaceId: row.id, error })
     }
   }
+
+  await Promise.all(stopping)
+
   if (failures.length > 0) {
     logger.error(`[${requestId}] Failed to archive workspaces for banned user ${userId}`, {
       failures: failures.map(({ workspaceId, error }) => ({
@@ -332,7 +380,7 @@ export async function disableUserResources(userId: string): Promise<void> {
   }
 
   logger.info(
-    `[${requestId}] Disabled resources for user ${userId}: archived ${ownedWorkspaces.length} workspaces, deleted API keys`
+    `[${requestId}] Disabled resources for user ${userId}: archived ${ownedWorkspaces.length} workspaces, deleted API keys, ${stopFailed ? 'failed to stop some of their runs' : 'stopped their runs'}`
   )
 }
 
diff --git a/packages/testing/src/mocks/async-jobs.mock.ts b/packages/testing/src/mocks/async-jobs.mock.ts
index d7ea9872872..863662b8da4 100644
--- a/packages/testing/src/mocks/async-jobs.mock.ts
+++ b/packages/testing/src/mocks/async-jobs.mock.ts
@@ -37,6 +37,7 @@ export const mockJobQueue = {
   cancelJob: vi.fn(),
   cancelByExecution: vi.fn(),
   cancelByKey: vi.fn(),
+  cancelByWorkspace: vi.fn(),
 }
 
 /**