@@ -19,8 +19,9 @@ import {
   createSessionPrincipal,
   createWorkspaceApiKeyPrincipal,
 } from '@sim/testing/factories/principal.factory'
-import { beforeEach, describe, expect, it, vi } from 'vitest'
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
 import { ADMISSION_ERROR_CODE } from '@/lib/core/admission/transient-failure'
+import { ExecutionScopeDeletedError } from '@/lib/core/errors/execution-scope-deleted'
 
 const {
   mockExecuteWorkflowCore,
@@ -440,6 +441,103 @@ describe('async preprocessing correlation threading', () => {
     })
   })
 
+  describe('a run whose workspace was deleted', () => {
+    afterEach(() => {
+      loggingSessionMockFns.mockGetDeletedScope.mockReset()
+      loggingSessionMockFns.mockSafeCompleteWithError.mockReset()
+    })
+
+    function runInDeletedWorkspace(executionId: string) {
+      mockPreprocessExecution.mockResolvedValueOnce({
+        success: true,
+        actorUserId: 'actor-1',
+        workflowRecord: {
+          id: 'workflow-1',
+          userId: 'owner-1',
+          workspaceId: 'workspace-1',
+          variables: {},
+        },
+        billingAttribution,
+        executionTimeout: {},
+      })
+      return executeWorkflowJob({
+        principal,
+        workflowId: 'workflow-1',
+        userId: 'actor-1',
+        workspaceId: 'workspace-1',
+        billingAttribution,
+        triggerType: 'api',
+        executionId,
+        requestId: 'request-deleted',
+      })
+    }
+
+    it('completes without alerting when core finds the workspace gone at start', async () => {
+      mockExecuteWorkflowCore.mockRejectedValueOnce(
+        new ExecutionScopeDeletedError('execution-deleted-start', 'workspace')
+      )
+
+      await expect(runInDeletedWorkspace('execution-deleted-start')).resolves.toMatchObject({
+        success: false,
+        executionId: 'execution-deleted-start',
+        error: 'The workspace of execution execution-deleted-start was deleted',
+      })
+      expect(loggingSessionMockFns.mockSafeCompleteWithError).not.toHaveBeenCalled()
+      expect(workflowExecutionLogger.info).toHaveBeenCalledWith(
+        '[request-deleted] Workflow execution ended: its workspace was deleted',
+        { workflowId: 'workflow-1', executionId: 'execution-deleted-start' }
+      )
+    })
+
+    it('completes without alerting when a block fails after the workspace is deleted mid-run', async () => {
+      const blockError = new Error('insert or update on table "workspace_files" violates FK')
+      mockExecuteWorkflowCore.mockRejectedValueOnce(blockError)
+      let deletedScope: 'workspace' | null = null
+      loggingSessionMockFns.mockGetDeletedScope.mockImplementation(() => deletedScope)
+      loggingSessionMockFns.mockWaitForPostExecution.mockImplementationOnce(async () => {
+        await Promise.resolve()
+        deletedScope = 'workspace'
+      })
+
+      await expect(runInDeletedWorkspace('execution-deleted-mid')).resolves.toMatchObject({
+        success: false,
+        executionId: 'execution-deleted-mid',
+        error: blockError.message,
+      })
+      expect(loggingSessionMockFns.mockSafeCompleteWithError).not.toHaveBeenCalled()
+    })
+
+    it('completes without alerting when its own failure log finds the workspace gone', async () => {
+      const pipelineError = new Error('Workflow state not found')
+      mockExecuteWorkflowCore.mockRejectedValueOnce(pipelineError)
+      loggingSessionMockFns.mockSafeCompleteWithError.mockImplementationOnce(async () => {
+        loggingSessionMockFns.mockGetDeletedScope.mockReturnValue('workspace')
+      })
+
+      await expect(runInDeletedWorkspace('execution-deleted-log')).resolves.toMatchObject({
+        success: false,
+        error: pipelineError.message,
+      })
+      expect(loggingSessionMockFns.mockSafeCompleteWithError).toHaveBeenCalledTimes(1)
+    })
+
+    it('completes without alerting when a run in a deleted workspace times out', async () => {
+      mockIsWorkflowTimedOut.mockReturnValue(true)
+      mockExecuteWorkflowCore.mockResolvedValueOnce({
+        success: false,
+        status: 'cancelled',
+        output: undefined,
+        metadata: { duration: 120_000, userId: 'actor-1' },
+      })
+      loggingSessionMockFns.mockGetDeletedScope.mockReturnValue('workspace')
+
+      await expect(runInDeletedWorkspace('execution-deleted-timeout')).resolves.toMatchObject({
+        success: false,
+        error: 'Execution timed out after 2 minutes',
+      })
+    })
+  })
+
   it('fails the backing job after a cooperative workflow timeout is finalized', async () => {
     mockPreprocessExecution.mockResolvedValueOnce({
       success: true,
 
@@ -24,7 +24,16 @@ import {
   workflowsPersistenceUtilsMockFns,
 } from '@sim/testing/mocks/workflows-persistence-utils.mock'
 import type { Mock } from 'vitest'
-import { afterAll, beforeEach, describe, expect, it, type MockInstance, vi } from 'vitest'
+import {
+  afterAll,
+  afterEach,
+  beforeEach,
+  describe,
+  expect,
+  it,
+  type MockInstance,
+  vi,
+} from 'vitest'
 import { getBlock } from '@/blocks'
 
 const {
@@ -119,6 +128,7 @@ vi.mock('@/tools/safe-assign', () => ({ safeAssign: vi.fn() }))
 vi.mock('@/triggers', () => triggersMock)
 
 import * as usageReservation from '@/lib/billing/calculations/usage-reservation'
+import { ExecutionScopeDeletedError } from '@/lib/core/errors/execution-scope-deleted'
 import { isRetryableSetupError } from '@/lib/core/errors/retryable-infrastructure'
 import { executeWebhookJob, type WebhookExecutionPayload } from '@/background/webhook-execution'
 
@@ -239,6 +249,7 @@ describe('executeWebhookJob fault vs error handling', () => {
         safeComplete: loggingSessionMockFns.mockSafeComplete,
         safeCompleteWithError: loggingSessionMockFns.mockSafeCompleteWithError,
         waitForPostExecution: loggingSessionMockFns.mockWaitForPostExecution,
+        getDeletedScope: loggingSessionMockFns.mockGetDeletedScope,
         markAsFailed: loggingSessionMockFns.mockMarkAsFailed,
         setExecutionDeadlineAt: loggingSessionMockFns.mockSetExecutionDeadlineAt,
         setResolvedSecretTraceRegistry: mockSetResolvedSecretTraceRegistry,
@@ -501,6 +512,55 @@ describe('executeWebhookJob fault vs error handling', () => {
     expect(rawError.message).toContain(secret)
   })
 
+  describe('a run whose workspace was deleted', () => {
+    afterEach(() => {
+      loggingSessionMockFns.mockGetDeletedScope.mockReset()
+    })
+
+    it('completes without alerting when core finds the workspace gone at start', async () => {
+      mockExecuteWorkflowCore.mockRejectedValue(
+        new ExecutionScopeDeletedError('execution-1', 'workspace')
+      )
+      mockWasExecutionFinalizedByCore.mockReturnValue(false)
+
+      await expect(executeWebhookJob(payload)).resolves.toMatchObject({
+        success: false,
+        executionId: 'execution-1',
+        provider: 'gmail',
+      })
+      expect(loggingSessionMockFns.mockSafeStart).toHaveBeenCalledTimes(0)
+      expect(loggingSessionMockFns.mockSafeCompleteWithError).not.toHaveBeenCalled()
+    })
+
+    it('completes without alerting when a block fails after the workspace is deleted mid-run', async () => {
+      mockExecuteWorkflowCore.mockRejectedValue(new Error('violates foreign key constraint'))
+      mockWasExecutionFinalizedByCore.mockReturnValue(false)
+      let deletedScope: 'workspace' | null = null
+      loggingSessionMockFns.mockGetDeletedScope.mockImplementation(() => deletedScope)
+      loggingSessionMockFns.mockWaitForPostExecution.mockImplementationOnce(async () => {
+        await Promise.resolve()
+        deletedScope = 'workspace'
+      })
+
+      await expect(executeWebhookJob(payload)).resolves.toMatchObject({
+        success: false,
+        executionId: 'execution-1',
+      })
+      expect(loggingSessionMockFns.mockSafeCompleteWithError).not.toHaveBeenCalled()
+    })
+
+    it('completes without alerting when its own failure log finds the workspace gone', async () => {
+      mockExecuteWorkflowCore.mockRejectedValue(new Error('Workflow state not found'))
+      mockWasExecutionFinalizedByCore.mockReturnValue(false)
+      loggingSessionMockFns.mockSafeStart.mockImplementationOnce(async () => {
+        loggingSessionMockFns.mockGetDeletedScope.mockReturnValue('workspace')
+        return false
+      })
+
+      await expect(executeWebhookJob(payload)).resolves.toMatchObject({ success: false })
+    })
+  })
+
   it('executes against the deployment version admitted by webhook ingress', async () => {
     mockExecuteWorkflowCore.mockResolvedValue({
       success: true,
 
@@ -25,6 +25,7 @@ import {
 import { getJobQueue } from '@/lib/core/async-jobs'
 import type { AsyncExecutionCorrelation } from '@/lib/core/async-jobs/types'
 import { env, envNumber } from '@/lib/core/config/env'
+import type { ExecutionScope } from '@/lib/core/errors/execution-scope-deleted'
 import { logFailureOnce } from '@/lib/core/errors/failure-log'
 import {
   describeRetryableInfrastructureError,
@@ -76,7 +77,10 @@ import {
   type PreloadedExecutionEnvironment,
 } from '@/lib/workflows/executor/execution-core'
 import { handlePostExecutionPauseState } from '@/lib/workflows/executor/pause-persistence'
-import { isRecordedWorkflowFailure } from '@/lib/workflows/executor/recorded-failure'
+import {
+  isRecordedWorkflowFailure,
+  resolveDeletedExecutionScope,
+} from '@/lib/workflows/executor/recorded-failure'
 import {
   loadDeployedWorkflowState,
   loadWorkflowDeploymentVersionState,
@@ -1277,19 +1281,33 @@ async function executeWebhookJobInternal(
       executionId,
     })
 
+    const failedRunResult = () => ({
+      success: false,
+      workflowId: payload.workflowId,
+      executionId,
+      output: hasExecutionResult(error) ? error.executionResult.output : {},
+      executedAt: new Date().toISOString(),
+      provider: payload.provider,
+    })
+    /** A run whose workspace or workflow was deleted has no owner left to alert and no log to write. */
+    const logDeletedScope = (scope: ExecutionScope) =>
+      logger.info(`[${requestId}] Webhook execution ended: its ${scope} was deleted`, {
+        workflowId: payload.workflowId,
+        executionId,
+      })
+
+    const deletedScope = await resolveDeletedExecutionScope(error, loggingSession)
+    if (deletedScope) {
+      logDeletedScope(deletedScope)
+      return failedRunResult()
+    }
+
     // A failure inside workflow execution (block error, provider 4xx, missing required field, etc.)
     // is finalized by core and already recorded in the execution logs. That is a user/workflow error,
     // not a trigger.dev job fault — complete the run normally so we don't fire a false alert. Errors
     // that were not finalized came from the webhook pipeline itself, so we re-throw to fault below.
     if (await isRecordedWorkflowFailure(error, executionId, loggingSession)) {
-      return {
-        success: false,
-        workflowId: payload.workflowId,
-        executionId,
-        output: hasExecutionResult(error) ? error.executionResult.output : {},
-        executedAt: new Date().toISOString(),
-        provider: payload.provider,
-      }
+      return failedRunResult()
     }
 
     try {
@@ -1332,6 +1350,12 @@ async function executeWebhookJobInternal(
       )
     }
 
+    const scopeDeletedOnCompletion = loggingSession.getDeletedScope()
+    if (scopeDeletedOnCompletion) {
+      logDeletedScope(scopeDeletedOnCompletion)
+      return failedRunResult()
+    }
+
     if (retryableSetupCause) {
       throw new RetryableSetupError(errorMessage, { cause: retryableSetupCause })
     }
 
@@ -18,6 +18,7 @@ import {
 } from '@/lib/billing/core/attribution'
 import { buildRecordedJobFailureResult } from '@/lib/core/async-jobs/recorded-failure'
 import type { AsyncExecutionCorrelation } from '@/lib/core/async-jobs/types'
+import type { ExecutionScope } from '@/lib/core/errors/execution-scope-deleted'
 import { logFailureOnce } from '@/lib/core/errors/failure-log'
 import {
   capExecutionTimeoutMs,
@@ -35,7 +36,10 @@ import { buildTraceSpans } from '@/lib/logs/execution/trace-spans/trace-spans'
 import { cleanupExecutionBase64Cache } from '@/lib/uploads/utils/user-file-base64.server'
 import { executeWorkflowCore } from '@/lib/workflows/executor/execution-core'
 import { handlePostExecutionPauseState } from '@/lib/workflows/executor/pause-persistence'
-import { isRecordedWorkflowFailure } from '@/lib/workflows/executor/recorded-failure'
+import {
+  isRecordedWorkflowFailure,
+  resolveDeletedExecutionScope,
+} from '@/lib/workflows/executor/recorded-failure'
 import { WORKFLOW_EXECUTION_CONCURRENCY_LIMIT } from '@/background/concurrency-limits'
 import { ExecutionSnapshot } from '@/executor/execution/snapshot'
 import type { ExecutionMetadata } from '@/executor/execution/types'
@@ -126,6 +130,24 @@ function completeWithRecordedFailure(workflowId: string, executionId: string, er
   return buildRecordedJobFailureResult({ workflowId, executionId, error: failure.message })
 }
 
+/**
+ * Completes, without alerting, a job whose workspace or workflow was deleted: no owner is left to
+ * act on an alert and no log is left to record the failure in.
+ */
+function completeWithDeletedScope(
+  requestId: string,
+  workflowId: string,
+  executionId: string,
+  scope: ExecutionScope,
+  error: unknown
+) {
+  logger.info(`[${requestId}] Workflow execution ended: its ${scope} was deleted`, {
+    workflowId,
+    executionId,
+  })
+  return completeWithRecordedFailure(workflowId, executionId, error)
+}
+
 function requireLegacyWorkflowJobString(value: unknown, field: string): string {
   if (typeof value !== 'string' || !value.trim()) {
     throw new Error(`Legacy workflow job ${field} must be a non-empty string`)
@@ -336,6 +358,12 @@ export async function executeWorkflowJob(
           executionId,
         })
 
+        // Checked before the timeout: a deleted scope leaves nobody to alert about a slow run.
+        const deletedScope = await resolveDeletedExecutionScope(error, loggingSession)
+        if (deletedScope) {
+          return completeWithDeletedScope(requestId, workflowId, executionId, deletedScope, error)
+        }
+
         if (error instanceof ExecutionTimeoutError) throw error
 
         // A failure core already recorded is the workflow's outcome, not a job fault; a run past
@@ -359,6 +387,17 @@ export async function executeWorkflowJob(
           executionState: executionResult?.executionState,
         })
 
+        const scopeDeletedOnCompletion = loggingSession.getDeletedScope()
+        if (scopeDeletedOnCompletion) {
+          return completeWithDeletedScope(
+            requestId,
+            workflowId,
+            executionId,
+            scopeDeletedOnCompletion,
+            error
+          )
+        }
+
         // Complete only when the rejection is durably logged; an unsaved one must still alert.
         if (
           preprocessingRejection &&
 
@@ -1,3 +1,7 @@
+import {
+  type ExecutionScope,
+  ExecutionScopeDeletedError,
+} from '@/lib/core/errors/execution-scope-deleted'
 import type { LoggingSession } from '@/lib/logs/execution/logging-session'
 import { wasExecutionFinalizedByCore } from '@/lib/workflows/executor/execution-core'
 
@@ -17,3 +21,19 @@ export async function isRecordedWorkflowFailure(
   await loggingSession.waitForPostExecution()
   return wasExecutionFinalizedByCore(error, executionId)
 }
+
+/**
+ * The run's workspace or workflow that was hard-deleted, or `null`. Core throws
+ * {@link ExecutionScopeDeletedError} when the scope is gone at start; a scope
+ * deleted mid-run surfaces only on the session once core's post-execution
+ * finalization fails to write the log, so this settles that promise first.
+ * Such a run has no owner left to alert and no log it could ever record.
+ */
+export async function resolveDeletedExecutionScope(
+  error: unknown,
+  loggingSession: Pick<LoggingSession, 'waitForPostExecution' | 'getDeletedScope'>
+): Promise<ExecutionScope | null> {
+  if (error instanceof ExecutionScopeDeletedError) return error.scope
+  await loggingSession.waitForPostExecution()
+  return loggingSession.getDeletedScope()
+}