@@ -55,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
@@ -424,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
@@ -538,7 +541,6 @@ export class DatabaseJobQueue implements JobQueueBackend {
       sql`${asyncJobs.metadata}->>'workspaceId' = ${workspaceId}`
     )
     let cancelledJobs = 0
-    let complete = false
     while (cancelledJobs < options.maxRuns && !options.signal?.aborted) {
       const batchSize = Math.min(WORKSPACE_CANCELLATION_BATCH_SIZE, options.maxRuns - cancelledJobs)
       const now = new Date()
@@ -556,21 +558,16 @@ export class DatabaseJobQueue implements JobQueueBackend {
         )
         .returning({ id: asyncJobs.id })
 
-      for (const { id } of cancelled) {
-        const controller = inlineAbortControllers.get(id)
-        if (!controller) continue
-        controller.abort(new DOMException('user', 'AbortError'))
-        inlineAbortControllers.delete(id)
-      }
+      for (const { id } of cancelled) abortInlineJob(id)
       cancelledJobs += cancelled.length
       if (cancelled.length < batchSize) {
-        complete = true
-        break
+        logger.info('Cancelled database jobs for workspace', { workspaceId, cancelledJobs })
+        return { cancelledJobs, complete: true }
       }
     }
 
-    logger.info('Cancelled database jobs for workspace', { workspaceId, cancelledJobs, complete })
-    return { cancelledJobs, complete }
+    logger.info('Paused database job sweep for workspace', { workspaceId, cancelledJobs })
+    return { cancelledJobs, complete: false }
   }
 
   cancelByKey(cancelKey: string): boolean {
 
@@ -128,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)}`
@@ -172,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
 }
 
@@ -186,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> {
@@ -200,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
@@ -232,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)
@@ -665,44 +673,33 @@ export class TriggerDevJobQueue implements JobQueueBackend {
     workspaceId: string,
     options: WorkspaceJobCancellationOptions
   ): Promise<WorkspaceJobCancellationResult> {
-    const workspaceTag = `workspaceId:${workspaceId}`
+    const workspaceTag = buildWorkspaceTag(workspaceId)
     const state: CancellationScanState = { cancelledJobs: 0, failureCount: 0 }
-    let pending: string[] = []
     let inspectedRuns = 0
     let complete = true
 
-    const flush = async (): Promise<void> => {
-      const runIds = pending
-      pending = []
-      const results = await Promise.allSettled(runIds.map((runId) => this.cancelJob(runId)))
-      for (const result of results) {
-        if (result.status === 'rejected') recordCancellationCandidateFailure(state, result.reason)
-        else state.cancelledJobs += 1
-      }
-    }
-
-    try {
-      for await (const run of runs.list({
-        tag: workspaceTag,
-        taskIdentifier: [...WORKSPACE_EXECUTION_JOB_TYPES],
-        status: [...ACTIVE_TRIGGER_RUN_STATUSES],
-        from: new Date(Date.now() - JOB_PENDING_RETENTION_HOURS * 60 * 60 * 1000),
-        limit: WORKSPACE_CANCELLATION_LIST_PAGE_SIZE,
-      })) {
-        if (inspectedRuns >= options.maxRuns || options.signal?.aborted) {
-          complete = false
-          break
-        }
-        if (!run.tags.includes(workspaceTag)) continue
+    await scanAndCancelTriggerRuns({
+      cancelJob: (jobId) => this.cancelJob(jobId),
+      listRuns: () =>
+        runs.list({
+          tag: workspaceTag,
+          taskIdentifier: [...WORKSPACE_EXECUTION_JOB_TYPES],
+          status: [...ACTIVE_TRIGGER_RUN_STATUSES],
+          from: new Date(Date.now() - JOB_PENDING_RETENTION_HOURS * 60 * 60 * 1000),
+          limit: WORKSPACE_CANCELLATION_LIST_PAGE_SIZE,
+        }),
+      shouldStop: () => {
+        if (inspectedRuns < options.maxRuns && !options.signal?.aborted) return false
+        complete = false
+        return true
+      },
+      selectCandidate: (run) => {
+        if (!run.tags.includes(workspaceTag)) return undefined
         inspectedRuns += 1
-        pending.push(run.id)
-        if (pending.length >= CANCELLATION_OPERATION_CHUNK_SIZE) await flush()
-      }
-    } catch (error) {
-      recordCancellationCandidateFailure(state, error)
-    } finally {
-      await flush()
-    }
+        return { id: run.id, verifyPayload: false }
+      },
+      state,
+    })
 
     if (state.firstError) {
       logger.error('Failed to cancel trigger.dev runs for workspace', {
@@ -748,7 +745,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}`)
 
 
@@ -43,7 +43,7 @@ export interface ExecutionStopPlan {
   executions: PlannedExecutionStop[]
 }
 
-export interface ExecutionStopPayload {
+interface ExecutionStopPayload {
   /** Workspace whose queued and running jobs this event sweeps. */
   workspaceId?: string
   executions: { executionId: string; executionDeadlineAt: string | null }[]
@@ -89,7 +89,6 @@ export async function collectExecutionStops(
     })
     .from(workflowExecutionLogs)
     .where(and(eq(workflowExecutionLogs.status, 'running'), inScope))
-  /** Driven from the scope's log rows, so `resume_queue` is read through its parent index. */
   const resuming = await executor
     .select({
       executionId: resumeQueue.newExecutionId,
@@ -142,17 +141,14 @@ function toPayloadExecutions(
 
 /** One event per removed workspace carries its sweep; executions beyond one event's share spill over. */
 function buildPayloads(plan: ExecutionStopPlan): ExecutionStopPayload[] {
-  const removedWorkspaces = new Set(plan.workspaceIds)
   const byWorkspace = new Map<string, PlannedExecutionStop[]>(
     plan.workspaceIds.map((workspaceId) => [workspaceId, []])
   )
   const elsewhere: PlannedExecutionStop[] = []
   for (const execution of plan.executions) {
-    if (execution.workspaceId && removedWorkspaces.has(execution.workspaceId)) {
-      byWorkspace.get(execution.workspaceId)?.push(execution)
-    } else {
-      elsewhere.push(execution)
-    }
+    const removed = execution.workspaceId ? byWorkspace.get(execution.workspaceId) : undefined
+    if (removed) removed.push(execution)
+    else elsewhere.push(execution)
   }
 
   const payloads: ExecutionStopPayload[] = []
@@ -174,12 +170,10 @@ function buildPayloads(plan: ExecutionStopPlan): ExecutionStopPayload[] {
 export async function enqueueExecutionStops(
   executor: Pick<typeof db, 'insert'>,
   plan: ExecutionStopPlan
-): Promise<string[]> {
-  const eventIds: string[] = []
+): Promise<void> {
   for (const chunk of chunkArray(buildPayloads(plan), MAX_BULK_ENQUEUE_EVENTS)) {
-    eventIds.push(...(await enqueueOutboxEvents(executor, EXECUTION_REMOVAL_STOP_EVENT, chunk)))
+    await enqueueOutboxEvents(executor, EXECUTION_REMOVAL_STOP_EVENT, chunk)
   }
-  return eventIds
 }
 
 /**
 
@@ -1,5 +1,5 @@
 import { db } from '@sim/db'
-import { isRecordLike } from '@sim/utils/object'
+import { z } from 'zod'
 import { getInlineJobQueue, getJobQueue } from '@/lib/core/async-jobs'
 import {
   continueOutboxHandler,
@@ -8,40 +8,24 @@ import {
   withOutboxHandlerTimeout,
 } from '@/lib/core/outbox/service'
 import { EXECUTION_REMOVAL_STOP_EVENT } from '@/lib/execution/removal-stop-event'
-import {
-  collectExecutionStops,
-  type ExecutionStopPayload,
-  signalExecutionStops,
-} from '@/lib/execution/removal-stop-records'
+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
 
-function parsePayload(payload: unknown): ExecutionStopPayload {
-  if (!isRecordLike(payload) || !Array.isArray(payload.executions)) {
-    throw new Error('Execution removal stop payload is malformed')
-  }
-  let workspaceId: string | undefined
-  if (payload.workspaceId !== undefined) {
-    if (typeof payload.workspaceId !== 'string' || payload.workspaceId.length === 0) {
-      throw new Error('Execution removal stop payload has an invalid workspaceId')
-    }
-    workspaceId = payload.workspaceId
-  }
-  const executions = payload.executions.map((entry: unknown) => {
-    if (
-      !isRecordLike(entry) ||
-      typeof entry.executionId !== 'string' ||
-      entry.executionId.length === 0 ||
-      (entry.executionDeadlineAt !== null && typeof entry.executionDeadlineAt !== 'string')
-    ) {
-      throw new Error('Execution removal stop payload has an invalid execution')
-    }
-    return { executionId: entry.executionId, executionDeadlineAt: entry.executionDeadlineAt }
-  })
-  return { ...(workspaceId ? { workspaceId } : {}), executions }
-}
+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
@@ -70,15 +54,9 @@ async function sweepWorkspaceJobs(workspaceId: string, signal: AbortSignal) {
  */
 const stopRemovedExecutions: OutboxHandler<unknown> = withOutboxHandlerTimeout(
   async (rawPayload, context) => {
-    const payload = parsePayload(rawPayload)
+    const payload = payloadSchema.parse(rawPayload)
     const executions = new Map(
-      payload.executions.map(({ executionId, executionDeadlineAt }) => [
-        executionId,
-        {
-          executionId,
-          executionDeadlineAt: executionDeadlineAt ? new Date(executionDeadlineAt) : null,
-        },
-      ])
+      payload.executions.map((execution) => [execution.executionId, execution])
     )
     /**
      * An archived workspace keeps its rows, so a run that passed admission before the archive
@@ -90,10 +68,12 @@ const stopRemovedExecutions: OutboxHandler<unknown> = withOutboxHandlerTimeout(
       })
       for (const execution of live) executions.set(execution.executionId, execution)
     }
-    const unrecorded = await signalExecutionStops([...executions.values()], context.signal)
-    const sweep = payload.workspaceId
-      ? await sweepWorkspaceJobs(payload.workspaceId, context.signal)
-      : { cancelledJobs: 0, complete: true }
+    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`)
     }