diff --git a/apps/sim/app/api/webhooks/trigger/[path]/route.ts b/apps/sim/app/api/webhooks/trigger/[path]/route.ts
index 0a8a90ca774..5c0999db993 100644
--- a/apps/sim/app/api/webhooks/trigger/[path]/route.ts
+++ b/apps/sim/app/api/webhooks/trigger/[path]/route.ts
@@ -141,6 +141,13 @@ async function handleWebhookDelivery(
     ? Number(slackRequestTimestamp) * 1000
     : undefined
 
+  /**
+   * Depends only on the path, so the read-only lookup overlaps the body stream
+   * read below; a challenge short-circuit simply abandons the result.
+   */
+  const webhookLookupPromise = findAllWebhooksForPath({ requestId, path })
+  webhookLookupPromise.catch(() => {})
+
   const parseResult = await parseWebhookBody(request, requestId)
 
   // Check if parseWebhookBody returned an error response
@@ -159,8 +166,8 @@ async function handleWebhookDelivery(
     return challengeResponse
   }
 
-  // Find all webhooks for this path (multiple webhooks in one workflow may share a path)
-  const allWebhooksForPath = await findAllWebhooksForPath({ requestId, path })
+  // Multiple webhooks in one workflow may share a path
+  const allWebhooksForPath = await webhookLookupPromise
 
   const pathWebhooks = allWebhooksForPath.filter(({ webhook: foundWebhook }) =>
     acceptsPathWebhookDelivery(foundWebhook.provider)
diff --git a/apps/sim/background/webhook-execution.test.ts b/apps/sim/background/webhook-execution.test.ts
index 822e908413a..e3fc760bf42 100644
--- a/apps/sim/background/webhook-execution.test.ts
+++ b/apps/sim/background/webhook-execution.test.ts
@@ -365,6 +365,216 @@ describe('executeWebhookJob fault vs error handling', () => {
     )
   })
 
+  it('reuses ingest-loaded rows and skips duplicate account checks with warm context', async () => {
+    mockExecuteWorkflowCore.mockResolvedValue({
+      success: true,
+      status: 'completed',
+      output: {},
+      logs: [],
+      executionState: {
+        blockStates: {},
+        executedBlocks: [],
+        blockLogs: [],
+        decisions: {},
+        completedLoops: [],
+        activeExecutionPath: [],
+      },
+    })
+    const warmContext = {
+      workflowRecord: { id: 'workflow-1', workspaceId: 'workspace-1', userId: 'user-1' },
+      webhookRecord: { id: 'webhook-1', providerConfig: { warm: true } },
+    } as unknown as NonNullable<Parameters<typeof executeWebhookJob>[2]>
+
+    await executeWebhookJob(payload, undefined, warmContext)
+
+    expect(executionPreprocessingMockFns.mockPreprocessExecution).toHaveBeenCalledWith(
+      expect.objectContaining({
+        workflowRecord: expect.objectContaining({ id: 'workflow-1' }),
+        trustWorkflowRecord: true,
+        skipAccountChecks: true,
+      })
+    )
+    expect(dbChainMockFns.limit).not.toHaveBeenCalled()
+    expect(mockResolveWebhookRecordProviderConfig).toHaveBeenCalledWith(
+      expect.objectContaining({ id: 'webhook-1', providerConfig: { warm: true } }),
+      'user-1',
+      'workspace-1',
+      expect.any(Object)
+    )
+  })
+
+  it('forwards the resolved credential owner into formatInput when the credential matches', async () => {
+    const formatInput = vi.fn().mockResolvedValue({ input: { event: {} } })
+    mockGetProviderHandler.mockReturnValue({ formatInput })
+    const { resolveOAuthAccountId } = await import('@/lib/oauth/credential-service')
+    vi.mocked(resolveOAuthAccountId).mockResolvedValue({
+      accountId: 'account-1',
+    } as never)
+    dbChainMockFns.limit.mockResolvedValue([{ userId: 'owner-1' }])
+    mockExecuteWorkflowCore.mockResolvedValue({
+      success: true,
+      status: 'completed',
+      output: {},
+      logs: [],
+      executionState: {
+        blockStates: {},
+        executedBlocks: [],
+        blockLogs: [],
+        decisions: {},
+        completedLoops: [],
+        activeExecutionPath: [],
+      },
+    })
+    const warmContext = {
+      workflowRecord: { id: 'workflow-1', workspaceId: 'workspace-1', userId: 'user-1' },
+      webhookRecord: { id: 'webhook-1', providerConfig: { credentialId: 'credential-1' } },
+    } as unknown as NonNullable<Parameters<typeof executeWebhookJob>[2]>
+
+    await executeWebhookJob({ ...payload, credentialId: 'credential-1' }, undefined, warmContext)
+
+    expect(formatInput).toHaveBeenCalledWith(
+      expect.objectContaining({ credentialOwnerUserId: 'owner-1' })
+    )
+  })
+
+  it('forwards the payload syncInteraction into formatInput', async () => {
+    const formatInput = vi.fn().mockResolvedValue({ input: { event: {} } })
+    mockGetProviderHandler.mockReturnValue({ formatInput })
+    mockExecuteWorkflowCore.mockResolvedValue({
+      success: true,
+      status: 'completed',
+      output: {},
+      logs: [],
+      executionState: {
+        blockStates: {},
+        executedBlocks: [],
+        blockLogs: [],
+        decisions: {},
+        completedLoops: [],
+        activeExecutionPath: [],
+      },
+    })
+
+    await executeWebhookJob({
+      ...payload,
+      syncInteraction: { loadingViewId: 'V-loading' },
+    })
+
+    expect(formatInput).toHaveBeenCalledWith(
+      expect.objectContaining({ syncInteraction: { loadingViewId: 'V-loading' } })
+    )
+  })
+
+  it('loads rows and keeps account checks without warm context', async () => {
+    mockExecuteWorkflowCore.mockResolvedValue({
+      success: true,
+      status: 'completed',
+      output: {},
+      logs: [],
+      executionState: {
+        blockStates: {},
+        executedBlocks: [],
+        blockLogs: [],
+        decisions: {},
+        completedLoops: [],
+        activeExecutionPath: [],
+      },
+    })
+
+    await executeWebhookJob(payload)
+
+    expect(executionPreprocessingMockFns.mockPreprocessExecution).toHaveBeenCalledWith(
+      expect.objectContaining({
+        workflowRecord: undefined,
+        trustWorkflowRecord: false,
+        skipAccountChecks: false,
+      })
+    )
+    expect(dbChainMockFns.limit).toHaveBeenCalled()
+  })
+
+  it('ignores warm rows whose ids do not match the payload', async () => {
+    mockExecuteWorkflowCore.mockResolvedValue({
+      success: true,
+      status: 'completed',
+      output: {},
+      logs: [],
+      executionState: {
+        blockStates: {},
+        executedBlocks: [],
+        blockLogs: [],
+        decisions: {},
+        completedLoops: [],
+        activeExecutionPath: [],
+      },
+    })
+    const warmContext = {
+      workflowRecord: { id: 'other-workflow' },
+      webhookRecord: { id: 'other-webhook' },
+    } as unknown as NonNullable<Parameters<typeof executeWebhookJob>[2]>
+
+    await executeWebhookJob(payload, undefined, warmContext)
+
+    expect(executionPreprocessingMockFns.mockPreprocessExecution).toHaveBeenCalledWith(
+      expect.objectContaining({
+        workflowRecord: undefined,
+        trustWorkflowRecord: false,
+        skipAccountChecks: false,
+      })
+    )
+    expect(dbChainMockFns.limit).toHaveBeenCalled()
+  })
+
+  it('logs phase timings and the executor-start metric for latency-tracked payloads', async () => {
+    mockExecuteWorkflowCore.mockResolvedValue({
+      success: true,
+      status: 'completed',
+      output: {},
+      logs: [],
+      executionState: {
+        blockStates: {},
+        executedBlocks: [],
+        blockLogs: [],
+        decisions: {},
+        completedLoops: [],
+        activeExecutionPath: [],
+      },
+    })
+
+    await executeWebhookJob({
+      ...payload,
+      webhookReceivedAt: Date.now() - 100,
+      triggerTimestampMs: Date.now() - 500,
+    })
+
+    expect(webhookExecutionLogger.info).toHaveBeenCalledWith(
+      '[request-1] Webhook dispatch latency',
+      expect.objectContaining({
+        dispatchLatencyMs: expect.any(Number),
+        triggerAgeMs: expect.any(Number),
+        preprocessMs: expect.any(Number),
+        loadsMs: expect.any(Number),
+        providerConfigMs: expect.any(Number),
+        formatInputMs: expect.any(Number),
+      })
+    )
+
+    const coreOptions = mockExecuteWorkflowCore.mock.calls[0]?.[0] as {
+      callbacks: { onBlockStart?: () => Promise<void> }
+    }
+    expect(coreOptions.callbacks.onBlockStart).toBeTypeOf('function')
+    await coreOptions.callbacks.onBlockStart?.()
+    await coreOptions.callbacks.onBlockStart?.()
+    const executorStartCalls = webhookExecutionLogger.info.mock.calls.filter(
+      ([message]: [string]) => String(message).includes('Webhook executor started')
+    )
+    expect(executorStartCalls).toHaveLength(1)
+    expect(executorStartCalls[0]?.[1]).toMatchObject({
+      executorStartLatencyMs: expect.any(Number),
+      executorStartTriggerAgeMs: expect.any(Number),
+    })
+  })
+
   it('does not pass provider-config provenance absent from the trigger input', async () => {
     mockGetEffectiveEnvironmentSnapshot.mockResolvedValue({
       personalEncrypted: { WEBHOOK_SECRET: 'personal-ciphertext' },
diff --git a/apps/sim/background/webhook-execution.ts b/apps/sim/background/webhook-execution.ts
index c5e6c79bbe9..89c928ed7e7 100644
--- a/apps/sim/background/webhook-execution.ts
+++ b/apps/sim/background/webhook-execution.ts
@@ -1,5 +1,5 @@
 import { db } from '@sim/db'
-import { account, webhook } from '@sim/db/schema'
+import { account, webhook, type workflow } from '@sim/db/schema'
 import { createLogger, runWithRequestContext } from '@sim/logger'
 import { toError } from '@sim/utils/errors'
 import { interruptibleSleep } from '@sim/utils/helpers'
@@ -56,6 +56,7 @@ import {
   type WebhookEnvResolutionOptions,
 } from '@/lib/webhooks/env-resolver'
 import { getProviderHandler } from '@/lib/webhooks/providers'
+import type { SyncInteractionContext } from '@/lib/webhooks/providers/types'
 import {
   executeWorkflowCore,
   wasExecutionFinalizedByCore,
@@ -306,6 +307,12 @@ export type WebhookExecutionPayload = {
    * first delivery and on legacy queued jobs.
    */
   infraRetryCount?: number
+  /**
+   * Interaction context created synchronously at ingest (e.g. a Slack loading
+   * modal's view id). Identifiers only — never token material; this payload is
+   * persisted by the durable queue branch.
+   */
+  syncInteraction?: SyncInteractionContext
 }
 
 const WEBHOOK_INFRA_RETRY_BASE_MS = envNumber(env.WEBHOOK_INFRA_RETRY_BASE_MS, 30_000, {
@@ -467,9 +474,21 @@ async function recordSetupFailureWithoutRequeue(
   }
 }
 
+/**
+ * Memory-only rows the same-process inline runner hands over so the worker does
+ * not re-read what ingest just loaded. Never serialized into the persisted job
+ * payload — Trigger.dev and recovery paths run without it and load everything
+ * themselves.
+ */
+export interface WebhookWarmContext {
+  workflowRecord?: typeof workflow.$inferSelect
+  webhookRecord?: typeof webhook.$inferSelect
+}
+
 export async function executeWebhookJob(
   payload: WebhookExecutionPayload,
-  externalAbortSignal?: AbortSignal
+  externalAbortSignal?: AbortSignal,
+  warmContext?: WebhookWarmContext
 ) {
   const correlation = buildWebhookCorrelation(payload)
   const executionId = correlation.executionId
@@ -534,7 +553,8 @@ export async function executeWebhookJob(
           payload,
           correlation,
           timeoutController,
-          admissionCompleted
+          admissionCompleted,
+          warmContext
         )
       }
 
@@ -682,7 +702,8 @@ async function executeWebhookJobInternal(
   payload: WebhookExecutionPayload,
   correlation: AsyncExecutionCorrelation,
   timeoutController: ReturnType<typeof createTimeoutAbortController>,
-  admissionCompleted: boolean
+  admissionCompleted: boolean,
+  warmContext?: WebhookWarmContext
 ) {
   const { executionId, requestId } = correlation
   const loggingSession = new LoggingSession(
@@ -693,6 +714,9 @@ async function executeWebhookJobInternal(
   )
   loggingSession.setExecutionDeadlineAt(getExecutionDeadlineAt(timeoutController.signal))
 
+  const warmWorkflowRecord =
+    warmContext?.workflowRecord?.id === payload.workflowId ? warmContext.workflowRecord : undefined
+  const preprocessStartedAt = Date.now()
   const preprocessResult = await preprocessExecution({
     workflowId: payload.workflowId,
     userId: payload.userId,
@@ -709,7 +733,11 @@ async function executeWebhookJobInternal(
     billingAttribution: payload.billingAttribution,
     executionType: 'async',
     executionDeadlineAt: getExecutionDeadlineAt(timeoutController.signal)?.getTime(),
+    workflowRecord: warmWorkflowRecord,
+    trustWorkflowRecord: Boolean(warmWorkflowRecord),
+    skipAccountChecks: admissionCompleted && Boolean(warmWorkflowRecord),
   })
+  const preprocessEndedAt = Date.now()
 
   if (!preprocessResult.success) {
     const failure = preprocessResult.error
@@ -772,19 +800,25 @@ async function executeWebhookJobInternal(
           workspaceId
         )
       : loadDeployedWorkflowState(payload.workflowId, workspaceId)
-    const [workflowData, webhookRows, resolvedCredentialUserId] = await Promise.all([
+    const warmWebhookRecord =
+      warmContext?.webhookRecord?.id === payload.webhookId ? warmContext.webhookRecord : undefined
+    /**
+     * Started here, awaited only where the owner id is first needed (formatInput),
+     * so its two serial reads overlap the state load and provider-config
+     * resolution instead of gating them. The empty catch marks the chain observed
+     * for the gap; the later await still surfaces the real error.
+     */
+    const credentialAccountUserIdPromise = payload.credentialId
+      ? resolveCredentialAccountUserId(payload.credentialId)
+      : Promise.resolve(undefined)
+    credentialAccountUserIdPromise.catch(() => {})
+    const [workflowData, webhookRows] = await Promise.all([
       workflowStatePromise,
-      db.select().from(webhook).where(eq(webhook.id, payload.webhookId)).limit(1),
-      payload.credentialId
-        ? resolveCredentialAccountUserId(payload.credentialId)
-        : Promise.resolve(undefined),
+      warmWebhookRecord
+        ? Promise.resolve([warmWebhookRecord])
+        : db.select().from(webhook).where(eq(webhook.id, payload.webhookId)).limit(1),
     ])
-    const credentialAccountUserId = resolvedCredentialUserId
-    if (payload.credentialId && !credentialAccountUserId) {
-      logger.warn(
-        `[${requestId}] Failed to resolve credential account for credential ${payload.credentialId}`
-      )
-    }
+    const loadsEndedAt = Date.now()
 
     if (!workflowData) {
       throw new Error(
@@ -842,6 +876,21 @@ async function executeWebhookJobInternal(
         },
       }
     )
+    const providerConfigEndedAt = Date.now()
+
+    const credentialAccountUserId = await credentialAccountUserIdPromise
+    if (payload.credentialId && !credentialAccountUserId) {
+      logger.warn(
+        `[${requestId}] Failed to resolve credential account for credential ${payload.credentialId}`
+      )
+    }
+    const resolvedProviderConfig = resolvedWebhookRecord.providerConfig
+    const formatInputCredentialOwnerUserId =
+      credentialAccountUserId &&
+      payload.credentialId &&
+      resolvedProviderConfig.credentialId === payload.credentialId
+        ? credentialAccountUserId
+        : undefined
 
     if (handler.formatInput) {
       const result = await handler.formatInput({
@@ -852,12 +901,17 @@ async function executeWebhookJobInternal(
         query: payload.query ?? {},
         method: payload.method ?? '',
         requestId,
+        ...(formatInputCredentialOwnerUserId
+          ? { credentialOwnerUserId: formatInputCredentialOwnerUserId }
+          : {}),
+        ...(payload.syncInteraction ? { syncInteraction: payload.syncInteraction } : {}),
       })
       input = result.input as Record<string, unknown> | null
       skipMessage = result.skip?.message
     } else {
       input = payload.body as Record<string, unknown> | null
     }
+    const formatInputEndedAt = Date.now()
 
     if (!input && handler.handleEmptyInput) {
       const skipResult = handler.handleEmptyInput(requestId)
@@ -1007,6 +1061,10 @@ async function executeWebhookJobInternal(
           payload.webhookReceivedAt !== undefined ? now - payload.webhookReceivedAt : undefined,
         triggerAgeMs:
           payload.triggerTimestampMs !== undefined ? now - payload.triggerTimestampMs : undefined,
+        preprocessMs: preprocessEndedAt - preprocessStartedAt,
+        loadsMs: loadsEndedAt - preprocessEndedAt,
+        providerConfigMs: providerConfigEndedAt - loadsEndedAt,
+        formatInputMs: formatInputEndedAt - providerConfigEndedAt,
       })
     }
 
@@ -1018,10 +1076,34 @@ async function executeWebhookJobInternal(
       []
     )
 
+    /**
+     * The dispatch-latency line above fires before executeWorkflowCore, so it cannot
+     * see the core's own setup (custom-block gate, env resolution, logging start,
+     * serialization). This logs once, when the first block actually starts — the
+     * moment that decides a trigger_id-bound provider's race against its expiry.
+     */
+    let executorStartLogged = false
+    const logExecutorStart = async () => {
+      if (executorStartLogged) return
+      executorStartLogged = true
+      if (payload.webhookReceivedAt === undefined && payload.triggerTimestampMs === undefined) {
+        return
+      }
+      const now = Date.now()
+      logger.info(`[${requestId}] Webhook executor started`, {
+        workflowId: payload.workflowId,
+        provider: payload.provider,
+        executorStartLatencyMs:
+          payload.webhookReceivedAt !== undefined ? now - payload.webhookReceivedAt : undefined,
+        executorStartTriggerAgeMs:
+          payload.triggerTimestampMs !== undefined ? now - payload.triggerTimestampMs : undefined,
+      })
+    }
+
     workflowCoreStarted = true
     const executionResult = await executeWorkflowCore({
       snapshot,
-      callbacks: {},
+      callbacks: { onBlockStart: logExecutorStart },
       loggingSession,
       trustedInitialResolvedSecretTraceProvenance:
         resolvedSecretTraceRegistry.exportProvenanceForValue(triggerInput),
diff --git a/apps/sim/lib/execution/preprocessing.test.ts b/apps/sim/lib/execution/preprocessing.test.ts
index 03f8a83c1f8..56eaf1afa9e 100644
--- a/apps/sim/lib/execution/preprocessing.test.ts
+++ b/apps/sim/lib/execution/preprocessing.test.ts
@@ -145,6 +145,124 @@ describe('preprocessExecution deployment checks', () => {
   })
 })
 
+describe('preprocessExecution prefetched record trust and account-check skips', () => {
+  type PrefetchedRecord = NonNullable<Parameters<typeof preprocessExecution>[0]['workflowRecord']>
+  const prefetchedRecord = {
+    id: 'workflow-1',
+    userId: 'creator-1',
+    workspaceId: 'workspace-1',
+    isDeployed: true,
+    archivedAt: null,
+  } as unknown as PrefetchedRecord
+
+  beforeEach(() => {
+    vi.clearAllMocks()
+  })
+
+  it('skips the active-record re-read when the prefetched record is trusted', async () => {
+    const result = await preprocessExecution({
+      workflowId: 'workflow-1',
+      userId: 'user-1',
+      triggerType: 'webhook',
+      executionId: 'execution-1',
+      requestId: 'request-1',
+      checkRateLimit: false,
+      workflowRecord: prefetchedRecord,
+      trustWorkflowRecord: true,
+    })
+
+    expect(result.success).toBe(true)
+    expect(workflowAuthzMockFns.mockGetActiveWorkflowRecord).not.toHaveBeenCalled()
+  })
+
+  it('re-reads the active record for a prefetched record without trust', async () => {
+    const result = await preprocessExecution({
+      workflowId: 'workflow-1',
+      userId: 'user-1',
+      triggerType: 'webhook',
+      executionId: 'execution-1',
+      requestId: 'request-1',
+      checkRateLimit: false,
+      workflowRecord: prefetchedRecord,
+    })
+
+    expect(result.success).toBe(true)
+    expect(workflowAuthzMockFns.mockGetActiveWorkflowRecord).toHaveBeenCalledWith('workflow-1')
+  })
+
+  it('still rejects a trusted prefetched record that is archived', async () => {
+    const result = await preprocessExecution({
+      workflowId: 'workflow-1',
+      userId: 'user-1',
+      triggerType: 'webhook',
+      executionId: 'execution-1',
+      requestId: 'request-1',
+      checkRateLimit: false,
+      workflowRecord: {
+        ...prefetchedRecord,
+        archivedAt: new Date('2026-08-01T00:00:00.000Z'),
+      } as unknown as PrefetchedRecord,
+      trustWorkflowRecord: true,
+    })
+
+    expect(result).toEqual({
+      success: false,
+      error: { message: 'Workflow not found', statusCode: 404 },
+    })
+    expect(workflowAuthzMockFns.mockGetActiveWorkflowRecord).not.toHaveBeenCalled()
+  })
+
+  it('skips the ban and subscription reads when skipAccountChecks is set', async () => {
+    const result = await preprocessExecution({
+      workflowId: 'workflow-1',
+      userId: 'user-1',
+      triggerType: 'webhook',
+      executionId: 'execution-1',
+      requestId: 'request-1',
+      checkRateLimit: false,
+      skipAccountChecks: true,
+      workflowRecord: prefetchedRecord,
+      trustWorkflowRecord: true,
+    })
+
+    expect(result.success).toBe(true)
+    if (result.success) {
+      expect(result.actorSubscription).toBeNull()
+    }
+    expect(mockGetActivelyBannedUserIds).not.toHaveBeenCalled()
+    expect(getHighestPrioritySubscription).not.toHaveBeenCalled()
+  })
+
+  it('keeps the ban and subscription reads by default', async () => {
+    const result = await preprocessExecution({
+      workflowId: 'workflow-1',
+      userId: 'user-1',
+      triggerType: 'webhook',
+      executionId: 'execution-1',
+      requestId: 'request-1',
+      checkRateLimit: false,
+    })
+
+    expect(result.success).toBe(true)
+    expect(mockGetActivelyBannedUserIds).toHaveBeenCalled()
+    expect(getHighestPrioritySubscription).toHaveBeenCalled()
+  })
+
+  it('rejects skipAccountChecks combined with rate limiting', async () => {
+    await expect(
+      preprocessExecution({
+        workflowId: 'workflow-1',
+        userId: 'user-1',
+        triggerType: 'webhook',
+        executionId: 'execution-1',
+        requestId: 'request-1',
+        checkRateLimit: true,
+        skipAccountChecks: true,
+      })
+    ).rejects.toThrow('skipAccountChecks requires checkRateLimit: false')
+  })
+})
+
 describe('preprocessExecution correlation logging', () => {
   it('preserves trigger correlation when logging preprocessing failures', async () => {
     mockResolveSystemBillingAttribution.mockRejectedValueOnce(
diff --git a/apps/sim/lib/execution/preprocessing.ts b/apps/sim/lib/execution/preprocessing.ts
index c68dda80679..f8d7ae2ac31 100644
--- a/apps/sim/lib/execution/preprocessing.ts
+++ b/apps/sim/lib/execution/preprocessing.ts
@@ -97,8 +97,25 @@ export interface PreprocessExecutionOptions {
   triggerData?: SessionStartParams['triggerData']
   /** Use the authenticated user as actor for client executions and personal API keys. */
   useAuthenticatedUserAsActor?: boolean
-  /** Pre-fetched workflow row for caller context; preprocessing still re-checks active state. */
+  /**
+   * Pre-fetched workflow row for caller context; preprocessing re-checks active
+   * state unless `trustWorkflowRecord` is set.
+   */
   workflowRecord?: WorkflowRecord
+  /**
+   * Trust the prefetched `workflowRecord` as current and skip the archived-state
+   * re-read. Only for callers that fetched the row in the same request (webhook
+   * ingest, or a same-process worker handed the ingest row); the archived guard
+   * still runs on the provided row.
+   */
+  trustWorkflowRecord?: boolean
+  /**
+   * Skip the ban and subscription reads. Only for the same-process inline path
+   * where ingest ran the full admission gates milliseconds earlier; queued and
+   * recovery jobs must keep them. `actorSubscription` resolves null, so this
+   * requires `checkRateLimit: false` (the rate-limit gate consumes it).
+   */
+  skipAccountChecks?: boolean
   /**
    * Immutable attribution captured by an upstream execution boundary. Background
    * and resume paths pass this through so payer ownership cannot change while
@@ -190,12 +207,20 @@ export async function preprocessExecution(
     triggerData,
     useAuthenticatedUserAsActor = false,
     workflowRecord: prefetchedWorkflowRecord,
+    trustWorkflowRecord = false,
+    skipAccountChecks = false,
     billingAttribution: providedBillingAttribution,
     executionType = 'sync',
     requestedTimeoutSeconds,
     executionDeadlineAt,
   } = options
 
+  if (skipAccountChecks && checkRateLimit) {
+    throw new Error(
+      'skipAccountChecks requires checkRateLimit: false — the rate-limit gate consumes the subscription'
+    )
+  }
+
   /** Suppresses log rows when the caller surfaces preprocessing failures itself. */
   const recordPreprocessingError: typeof logPreprocessingError = (args) =>
     logPreprocessingErrors ? logPreprocessingError(args) : Promise.resolve()
@@ -283,7 +308,7 @@ export async function preprocessExecution(
         statusCode: 404,
       },
     }
-  } else {
+  } else if (!trustWorkflowRecord) {
     const activeWorkflow = await getActiveWorkflowRecord(workflowId)
     if (!activeWorkflow) {
       logger.warn(`[${requestId}] Workflow archived before execution started: ${workflowId}`)
@@ -448,6 +473,7 @@ export async function preprocessExecution(
   }
 
   const banCheck = (async (): Promise<GateFailure | null> => {
+    if (skipAccountChecks) return null
     /**
      * Blocks when the resolved actor, workflow owner, or caller-provided user
      * has an active ban or blocked email domain. Including the workflow owner
@@ -519,7 +545,9 @@ export async function preprocessExecution(
     }
   })()
 
-  const subscriptionFetch = getHighestPrioritySubscription(actorUserId)
+  const subscriptionFetch = skipAccountChecks
+    ? Promise.resolve(null)
+    : getHighestPrioritySubscription(actorUserId)
 
   /**
    * Returns the usage failure and reservation snapshot together so concurrent
diff --git a/apps/sim/lib/logs/execution/logger.ts b/apps/sim/lib/logs/execution/logger.ts
index 00d04a72b2d..30b22b49dc0 100644
--- a/apps/sim/lib/logs/execution/logger.ts
+++ b/apps/sim/lib/logs/execution/logger.ts
@@ -651,12 +651,21 @@ export class ExecutionLogger implements IExecutionLoggerService {
 
     execLog.debug('Starting workflow execution')
 
-    // Check if execution log already exists (idempotency check)
-    const existingLog = await execDb
-      .select()
-      .from(workflowExecutionLogs)
-      .where(eq(workflowExecutionLogs.executionId, executionId))
-      .limit(1)
+    /**
+     * The duplicate-execution probe and the snapshot upsert have no data
+     * dependency, so they run concurrently. On the duplicate path the extra
+     * snapshot write is an idempotent no-op for an unchanged state hash (its
+     * `(workflowId, stateHash)` conflict target), and an orphaned row from a
+     * changed hash is reclaimed by `cleanupOrphanedSnapshots`.
+     */
+    const [existingLog, snapshotResult] = await Promise.all([
+      execDb
+        .select()
+        .from(workflowExecutionLogs)
+        .where(eq(workflowExecutionLogs.executionId, executionId))
+        .limit(1),
+      snapshotService.createSnapshotWithDeduplication(workflowId, workflowState),
+    ])
 
     if (existingLog.length > 0) {
       execLog.debug('Execution log already exists, skipping duplicate INSERT (idempotent)')
@@ -691,11 +700,6 @@ export class ExecutionLogger implements IExecutionLoggerService {
       }
     }
 
-    const snapshotResult = await snapshotService.createSnapshotWithDeduplication(
-      workflowId,
-      workflowState
-    )
-
     const startTime = new Date()
 
     const [workflowLog] = await execDb
diff --git a/apps/sim/lib/webhooks/processor.test.ts b/apps/sim/lib/webhooks/processor.test.ts
index a3257514ac1..605a641fe4b 100644
--- a/apps/sim/lib/webhooks/processor.test.ts
+++ b/apps/sim/lib/webhooks/processor.test.ts
@@ -536,7 +536,11 @@ describe('webhook processor execution identity', () => {
     await options.runner?.({}, controller.signal)
     expect(mockExecuteWebhookJob).toHaveBeenCalledWith(
       expect.objectContaining({ executionId: 'generated-execution-id' }),
-      controller.signal
+      controller.signal,
+      expect.objectContaining({
+        workflowRecord: expect.objectContaining({ id: 'workflow-1' }),
+        webhookRecord: expect.objectContaining({ id: 'webhook-1' }),
+      })
     )
   })
 
@@ -564,6 +568,80 @@ describe('webhook processor execution identity', () => {
   })
 })
 
+describe('webhook processor prepareSyncDispatch', () => {
+  beforeEach(() => {
+    vi.clearAllMocks()
+    mockPreprocessExecution.mockResolvedValue({
+      success: true,
+      actorUserId: 'actor-user-1',
+      billingAttribution,
+      executionTimeout: { sync: 0, async: 120_000 },
+    })
+    mockEnqueue.mockResolvedValue('job-1')
+    mockGetInlineJobQueue.mockResolvedValue({ enqueue: mockEnqueue })
+    mockGetJobQueue.mockResolvedValue({ enqueue: mockEnqueue })
+    mockProviderHandler.current = {}
+    mockShouldExecuteInline.mockReturnValue(false)
+    mockGenerateId.mockReturnValue('generated-execution-id')
+    workflowsPersistenceUtilsMockFns.mockBlockExistsInDeployment.mockResolvedValue(true)
+  })
+
+  const dispatch = () =>
+    dispatchResolvedWebhookTarget(
+      makeWebhookRecord({
+        path: 'incoming/slack',
+        provider: 'slack',
+        providerConfig: { openLoadingModal: true },
+      }),
+      makeWorkflowRecord({}),
+      { type: 'block_actions', trigger_id: 'trigger-1' },
+      createMockRequest('POST', { type: 'block_actions' }) as NextRequest,
+      { requestId: 'request-1', path: 'incoming/slack' }
+    )
+
+  it('puts the hook result on the enqueued payload', async () => {
+    const prepareSyncDispatch = vi
+      .fn()
+      .mockResolvedValue({ syncInteraction: { loadingViewId: 'V1' } })
+    mockProviderHandler.current = { prepareSyncDispatch }
+
+    const result = await dispatch()
+
+    expect(result.outcome).toBe('queued')
+    expect(prepareSyncDispatch).toHaveBeenCalledWith(
+      expect.objectContaining({
+        body: { type: 'block_actions', trigger_id: 'trigger-1' },
+        workflow: expect.objectContaining({ id: 'workflow-1' }),
+        providerConfig: { openLoadingModal: true },
+        requestId: 'request-1',
+      })
+    )
+    expect(mockEnqueue.mock.calls[0]?.[1]).toMatchObject({
+      syncInteraction: { loadingViewId: 'V1' },
+    })
+  })
+
+  it('omits syncInteraction when the hook resolves null', async () => {
+    mockProviderHandler.current = { prepareSyncDispatch: vi.fn().mockResolvedValue(null) }
+
+    const result = await dispatch()
+
+    expect(result.outcome).toBe('queued')
+    expect(mockEnqueue.mock.calls[0]?.[1]).not.toHaveProperty('syncInteraction')
+  })
+
+  it('still queues when the hook throws', async () => {
+    mockProviderHandler.current = {
+      prepareSyncDispatch: vi.fn().mockRejectedValue(new Error('slack down')),
+    }
+
+    const result = await dispatch()
+
+    expect(result.outcome).toBe('queued')
+    expect(mockEnqueue.mock.calls[0]?.[1]).not.toHaveProperty('syncInteraction')
+  })
+})
+
 describe('polled webhook reservation ownership', () => {
   const foundWebhook = {
     id: 'webhook-1',
@@ -714,7 +792,11 @@ describe('polled webhook reservation ownership', () => {
     await options.runner?.({}, controller.signal)
     expect(mockExecuteWebhookJob).toHaveBeenCalledWith(
       expect.objectContaining({ executionId: 'generated-execution-id' }),
-      controller.signal
+      controller.signal,
+      expect.objectContaining({
+        workflowRecord: expect.objectContaining({ id: 'workflow-1' }),
+        webhookRecord: expect.objectContaining({ id: 'webhook-1' }),
+      })
     )
   })
 })
diff --git a/apps/sim/lib/webhooks/processor.ts b/apps/sim/lib/webhooks/processor.ts
index ca8c6694404..9a88f83923d 100644
--- a/apps/sim/lib/webhooks/processor.ts
+++ b/apps/sim/lib/webhooks/processor.ts
@@ -32,7 +32,7 @@ import {
   requiresPendingWebhookVerification,
 } from '@/lib/webhooks/pending-verification'
 import { getProviderHandler } from '@/lib/webhooks/providers'
-import type { WebhookProviderHandler } from '@/lib/webhooks/providers/types'
+import type { SyncInteractionContext, WebhookProviderHandler } from '@/lib/webhooks/providers/types'
 import { normalizeWebhookRegistrationPath } from '@/lib/webhooks/registration-identity'
 import { blockExistsInDeployment } from '@/lib/workflows/persistence/utils'
 import { SIM_TRIGGER_PROVIDER } from '@/lib/workspace-events/constants'
@@ -600,6 +600,7 @@ export async function checkWebhookPreprocessing(
       checkDeployment: true,
       workspaceId: foundWorkflow.workspaceId ?? undefined,
       workflowRecord: foundWorkflow,
+      trustWorkflowRecord: true,
       executionType: 'async',
     })
 
@@ -722,6 +723,35 @@ async function queueWebhookExecutionWithResult(
         provider: foundWebhook.provider,
         triggerType: 'webhook',
       } satisfies AsyncExecutionCorrelation)
+
+    /**
+     * Runs after event filters, the deployment-block check, and admission — so a
+     * filtered or rejected delivery never produces a side effect — and before the
+     * payload is assembled so the result rides into execution. A hook failure
+     * never blocks dispatch.
+     */
+    let syncInteraction: SyncInteractionContext | undefined
+    if (handler.prepareSyncDispatch) {
+      try {
+        const prepared = await handler.prepareSyncDispatch({
+          webhook: foundWebhook,
+          workflow: {
+            id: foundWorkflow.id,
+            userId: foundWorkflow.userId,
+            workspaceId: foundWorkflow.workspaceId,
+          },
+          body,
+          requestId: options.requestId,
+          providerConfig,
+        })
+        syncInteraction = prepared?.syncInteraction
+      } catch (error) {
+        logger.warn(`[${options.requestId}] prepareSyncDispatch failed; continuing dispatch`, {
+          provider: foundWebhook.provider,
+          error: toError(error).message,
+        })
+      }
+    }
     const payload = {
       webhookId: foundWebhook.id,
       workflowId: foundWorkflow.id,
@@ -749,6 +779,7 @@ async function queueWebhookExecutionWithResult(
       ...(options.executionTimeoutMs !== undefined
         ? { executionTimeoutMs: options.executionTimeoutMs }
         : {}),
+      ...(syncInteraction ? { syncInteraction } : {}),
     } satisfies WebhookExecutionPayload
 
     const shouldUseQueue = shouldUseDurableQueue(payload.provider, handler)
@@ -779,7 +810,10 @@ async function queueWebhookExecutionWithResult(
         },
         maxDurationSeconds,
         runner: (_queuedPayload: unknown, signal: AbortSignal) =>
-          executeWebhookJob(payload, signal),
+          executeWebhookJob(payload, signal, {
+            workflowRecord: foundWorkflow,
+            webhookRecord: foundWebhook,
+          }),
       })
       reservationTransferred = true
       logger.info(
@@ -866,7 +900,10 @@ export async function dispatchResolvedWebhookTarget(
   }
 
   if (webhookRecord.blockId) {
-    const blockExists = await blockExistsInDeployment(foundWorkflow.id, webhookRecord.blockId)
+    const blockExists = await blockExistsInDeployment(foundWorkflow.id, webhookRecord.blockId, {
+      deploymentVersionId: webhookRecord.deploymentVersionId,
+      workspaceId: foundWorkflow.workspaceId,
+    })
     if (!blockExists) {
       const verificationResponse = handlePreDeploymentVerification(webhookRecord, options.requestId)
       return {
@@ -952,7 +989,10 @@ export async function processPolledWebhookEvent(
   let reservationTransferred = false
   try {
     if (foundWebhook.blockId) {
-      const blockExists = await blockExistsInDeployment(foundWorkflow.id, foundWebhook.blockId)
+      const blockExists = await blockExistsInDeployment(foundWorkflow.id, foundWebhook.blockId, {
+        deploymentVersionId: foundWebhook.deploymentVersionId,
+        workspaceId: foundWorkflow.workspaceId,
+      })
       if (!blockExists) {
         logger.info(
           `[${requestId}] Trigger block ${foundWebhook.blockId} not found in deployment for workflow ${foundWorkflow.id}`
@@ -1067,7 +1107,10 @@ export async function processPolledWebhookEvent(
         },
         maxDurationSeconds,
         runner: (_queuedPayload: unknown, signal: AbortSignal) =>
-          executeWebhookJob(payload, signal),
+          executeWebhookJob(payload, signal, {
+            workflowRecord: foundWorkflow,
+            webhookRecord: foundWebhook,
+          }),
       })
       reservationTransferred = true
       logger.info(`[${requestId}] Queued ${provider} webhook execution ${jobId} via inline backend`)
diff --git a/apps/sim/lib/webhooks/providers/slack.test.ts b/apps/sim/lib/webhooks/providers/slack.test.ts
index 839f8807e56..70af5767de7 100644
--- a/apps/sim/lib/webhooks/providers/slack.test.ts
+++ b/apps/sim/lib/webhooks/providers/slack.test.ts
@@ -1,4 +1,15 @@
-import { describe, expect, it } from 'vitest'
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
+
+const { mockGetSlackBotCredential } = vi.hoisted(() => ({
+  mockGetSlackBotCredential: vi.fn(),
+}))
+
+vi.mock('@/lib/oauth/credential-service', () => ({
+  getSlackBotCredential: mockGetSlackBotCredential,
+  refreshAccessTokenIfNeeded: vi.fn(),
+  resolveOAuthAccountId: vi.fn(),
+}))
+
 import {
   handleSlackChallenge,
   resolveSlackEventKey,
@@ -595,3 +606,185 @@ describe('slackHandler.shouldSkipEvent (custom-app path)', () => {
     expect(slackHandler.shouldSkipEvent!(skipCtx({}, message))).toBe(false)
   })
 })
+
+describe('slackHandler prepareSyncDispatch', () => {
+  const fetchMock = vi.fn()
+
+  const syncCtx = (body: unknown, providerConfig: Record<string, unknown>) => ({
+    webhook: {},
+    workflow: { id: 'wf', userId: 'u' },
+    body,
+    requestId: 'slack-test',
+    providerConfig,
+  })
+
+  const blockActions = (overrides: Record<string, unknown> = {}) => ({
+    type: 'block_actions',
+    trigger_id: 'trigger-1',
+    user: { id: 'U1' },
+    actions: [{ action_id: 'a1', type: 'button' }],
+    ...overrides,
+  })
+
+  const enabledConfig = { openLoadingModal: true, botToken: 'xoxb-test' }
+
+  const slackOk = (viewId = 'V123') => ({
+    json: async () => ({ ok: true, view: { id: viewId } }),
+    status: 200,
+  })
+
+  beforeEach(() => {
+    vi.clearAllMocks()
+    vi.stubGlobal('fetch', fetchMock)
+  })
+
+  afterEach(() => {
+    vi.unstubAllGlobals()
+  })
+
+  it('opens the loading modal for a message button click and returns the view id', async () => {
+    fetchMock.mockResolvedValue(slackOk('V123'))
+
+    const result = await slackHandler.prepareSyncDispatch!(syncCtx(blockActions(), enabledConfig))
+
+    expect(result).toEqual({ syncInteraction: { loadingViewId: 'V123' } })
+    expect(fetchMock).toHaveBeenCalledTimes(1)
+    const [url, init] = fetchMock.mock.calls[0]
+    expect(url).toBe('https://slack.com/api/views.open')
+    expect(init.headers.Authorization).toBe('Bearer xoxb-test')
+    const requestBody = JSON.parse(init.body)
+    expect(requestBody.trigger_id).toBe('trigger-1')
+    expect(requestBody.view.callback_id).toBe('sim_loading_modal')
+    expect(requestBody.view.title.text).toBe('Working on it')
+    expect(requestBody.view.blocks[0].text.text).toBe('This will just take a moment…')
+  })
+
+  it('opens for message_action and shortcut payloads', async () => {
+    fetchMock.mockResolvedValue(slackOk())
+
+    await expect(
+      slackHandler.prepareSyncDispatch!(
+        syncCtx({ type: 'message_action', trigger_id: 't2' }, enabledConfig)
+      )
+    ).resolves.toEqual({ syncInteraction: { loadingViewId: 'V123' } })
+    await expect(
+      slackHandler.prepareSyncDispatch!(
+        syncCtx({ type: 'shortcut', trigger_id: 't3' }, enabledConfig)
+      )
+    ).resolves.toEqual({ syncInteraction: { loadingViewId: 'V123' } })
+  })
+
+  it('applies the configured title and text, truncating the title to 24 characters', async () => {
+    fetchMock.mockResolvedValue(slackOk())
+
+    await slackHandler.prepareSyncDispatch!(
+      syncCtx(blockActions(), {
+        ...enabledConfig,
+        loadingModalTitle: 'A very long modal title that overflows',
+        loadingModalText: 'Custom body',
+      })
+    )
+
+    const requestBody = JSON.parse(fetchMock.mock.calls[0][1].body)
+    expect(requestBody.view.title.text.length).toBeLessThanOrEqual(24)
+    expect(requestBody.view.blocks[0].text.text).toBe('Custom body')
+  })
+
+  it('never fires for ineligible payloads or when the option is off', async () => {
+    const cases: Array<[unknown, Record<string, unknown>]> = [
+      [blockActions(), { botToken: 'xoxb-test' }],
+      [blockActions({ view: { id: 'V-open' } }), enabledConfig],
+      [{ type: 'view_submission', trigger_id: 't' }, enabledConfig],
+      [{ type: 'view_closed', trigger_id: 't' }, enabledConfig],
+      [{ event: { type: 'message' }, type: 'event_callback' }, enabledConfig],
+      [{ command: '/run', trigger_id: 't' }, enabledConfig],
+      [blockActions({ trigger_id: undefined }), enabledConfig],
+      [blockActions({ trigger_id: '' }), enabledConfig],
+      ['not-an-object', enabledConfig],
+    ]
+
+    for (const [body, config] of cases) {
+      await expect(slackHandler.prepareSyncDispatch!(syncCtx(body, config))).resolves.toBeNull()
+    }
+    expect(fetchMock).not.toHaveBeenCalled()
+  })
+
+  it('returns null without calling Slack when no bot token resolves', async () => {
+    await expect(
+      slackHandler.prepareSyncDispatch!(syncCtx(blockActions(), { openLoadingModal: true }))
+    ).resolves.toBeNull()
+    expect(fetchMock).not.toHaveBeenCalled()
+  })
+
+  it('resolves the token from a custom-bot credential', async () => {
+    mockGetSlackBotCredential.mockResolvedValue({ botToken: 'xoxb-cred' })
+    fetchMock.mockResolvedValue(slackOk('V-cred'))
+
+    const result = await slackHandler.prepareSyncDispatch!(
+      syncCtx(blockActions(), { openLoadingModal: true, credentialId: 'credential-1' })
+    )
+
+    expect(result).toEqual({ syncInteraction: { loadingViewId: 'V-cred' } })
+    expect(fetchMock.mock.calls[0][1].headers.Authorization).toBe('Bearer xoxb-cred')
+  })
+
+  it('tolerates Slack rejections, network failures, and timeouts without throwing', async () => {
+    fetchMock.mockResolvedValueOnce({
+      json: async () => ({ ok: false, error: 'expired_trigger_id' }),
+      status: 200,
+    })
+    await expect(
+      slackHandler.prepareSyncDispatch!(syncCtx(blockActions(), enabledConfig))
+    ).resolves.toBeNull()
+
+    fetchMock.mockResolvedValueOnce({
+      json: async () => ({ ok: false, error: 'exchanged_trigger_id' }),
+      status: 200,
+    })
+    await expect(
+      slackHandler.prepareSyncDispatch!(syncCtx(blockActions(), enabledConfig))
+    ).resolves.toBeNull()
+
+    fetchMock.mockRejectedValueOnce(new Error('network down'))
+    await expect(
+      slackHandler.prepareSyncDispatch!(syncCtx(blockActions(), enabledConfig))
+    ).resolves.toBeNull()
+
+    fetchMock.mockRejectedValueOnce(
+      Object.assign(new Error('The operation was aborted'), { name: 'TimeoutError' })
+    )
+    await expect(
+      slackHandler.prepareSyncDispatch!(syncCtx(blockActions(), enabledConfig))
+    ).resolves.toBeNull()
+  })
+})
+
+describe('slackHandler formatInput - loading_view_id', () => {
+  const interactiveBody = {
+    type: 'block_actions',
+    trigger_id: 'trigger-1',
+    user: { id: 'U1', username: 'alice' },
+    actions: [{ action_id: 'a1', type: 'button', value: 'go' }],
+  }
+
+  it('surfaces the ingest-created loading view id on interactive payloads', async () => {
+    const { input } = await slackHandler.formatInput!({
+      ...ctx(interactiveBody),
+      syncInteraction: { loadingViewId: 'V123' },
+    })
+
+    expect(eventOf(input).loading_view_id).toBe('V123')
+  })
+
+  it('defaults to an empty loading_view_id without a sync interaction', async () => {
+    const { input } = await slackHandler.formatInput!(ctx(interactiveBody))
+    expect(eventOf(input).loading_view_id).toBe('')
+  })
+
+  it('keeps the empty default on Events API payloads', async () => {
+    const { input } = await slackHandler.formatInput!(
+      ctx({ team_id: 'T1', event: { type: 'app_mention', channel: 'C1', ts: '1.2' } })
+    )
+    expect(eventOf(input).loading_view_id).toBe('')
+  })
+})
diff --git a/apps/sim/lib/webhooks/providers/slack.ts b/apps/sim/lib/webhooks/providers/slack.ts
index 9830f7a5a9a..0bd3e17a704 100644
--- a/apps/sim/lib/webhooks/providers/slack.ts
+++ b/apps/sim/lib/webhooks/providers/slack.ts
@@ -5,6 +5,7 @@ import { safeCompare } from '@sim/security/compare'
 import { hmacSha256Hex } from '@sim/security/hmac'
 import { toError } from '@sim/utils/errors'
 import { isRecordLike } from '@sim/utils/object'
+import { truncate } from '@sim/utils/string'
 import { eq } from 'drizzle-orm'
 import { NextResponse } from 'next/server'
 import {
@@ -21,6 +22,8 @@ import type {
   EventFilterContext,
   FormatInputContext,
   FormatInputResult,
+  PrepareSyncDispatchContext,
+  PrepareSyncDispatchResult,
   WebhookProviderHandler,
 } from '@/lib/webhooks/providers/types'
 import { type SlackEventFilter, slackEventSupportsFilter } from '@/triggers/slack/shared'
@@ -60,6 +63,27 @@ const SLACK_INTERACTIVE_TYPES = new Set([
  */
 const SLACK_INTERACTION_EVENT_KEYS = new Set(['block_actions', 'view_submission'])
 
+/**
+ * Interactive payload types whose fresh trigger_id may open a NEW modal. An
+ * interaction already inside a modal (top-level `view` present) is excluded —
+ * the workflow should views.update the existing `view` id instead of stacking
+ * a second modal on it.
+ */
+const SLACK_LOADING_MODAL_TYPES = new Set([
+  'block_actions',
+  'interactive_message',
+  'message_action',
+  'shortcut',
+])
+const SLACK_LOADING_MODAL_CALLBACK_ID = 'sim_loading_modal'
+/** Slack caps modal titles at 24 characters and section plain_text at 3000. */
+const SLACK_MODAL_TITLE_MAX_CHARS = 24
+const SLACK_SECTION_TEXT_MAX_CHARS = 3000
+/** Bounds the pre-ack cost of the synchronous views.open call. */
+const SLACK_VIEWS_OPEN_TIMEOUT_MS = 2000
+const DEFAULT_LOADING_MODAL_TITLE = 'Working on it'
+const DEFAULT_LOADING_MODAL_TEXT = 'This will just take a moment…'
+
 interface SlackDownloadedFile {
   name: string
   data: string
@@ -94,6 +118,13 @@ interface SlackTriggerEvent {
   actions: unknown[]
   response_url: string
   trigger_id: string
+  /**
+   * View id of the loading modal opened synchronously at ingest when the
+   * trigger's "Open loading modal" option is on. Update it with Slack Update
+   * View (view ids never expire, unlike trigger_id's 3-second window). Empty
+   * when the option is off or the open failed.
+   */
+  loading_view_id: string
   callback_id: string
   api_app_id: string
   app_id: string
@@ -144,6 +175,7 @@ function createSlackEvent(): SlackTriggerEvent {
     actions: [],
     response_url: '',
     trigger_id: '',
+    loading_view_id: '',
     callback_id: '',
     api_app_id: '',
     app_id: '',
@@ -808,6 +840,132 @@ export function shouldSkipSlackTriggerEvent(
   return false
 }
 
+/**
+ * Resolves the bot token a Slack webhook can act with, across the three trigger
+ * backends: a pasted bot token (legacy `slack_webhook`), a reusable custom-bot
+ * credential's stored token, and a native `slack_app` OAuth credential. The
+ * OAuth credential resolves via its OWNER (not the execution actor in
+ * `workflow.userId`, who may not own the credential) so reaction-message text
+ * and file downloads work. `credentialOwnerUserId` short-circuits the
+ * credential → account → owner chain when the caller already resolved it.
+ */
+async function resolveSlackWebhookBotToken(
+  providerConfig: Record<string, unknown>,
+  requestId: string,
+  credentialOwnerUserId?: string
+): Promise<string | undefined> {
+  const pastedToken = providerConfig.botToken as string | undefined
+  if (pastedToken || typeof providerConfig.credentialId !== 'string') {
+    return pastedToken
+  }
+  const credentialId = providerConfig.credentialId
+
+  const botCredential = await getSlackBotCredential(credentialId)
+  if (botCredential?.botToken) {
+    return botCredential.botToken
+  }
+
+  let ownerUserId = credentialOwnerUserId
+  if (!ownerUserId) {
+    const resolved = await resolveOAuthAccountId(credentialId)
+    if (!resolved?.accountId) {
+      return undefined
+    }
+    const [owner] = await db
+      .select({ userId: account.userId })
+      .from(account)
+      .where(eq(account.id, resolved.accountId))
+      .limit(1)
+    ownerUserId = owner?.userId
+  }
+  if (!ownerUserId) {
+    return undefined
+  }
+  return (await refreshAccessTokenIfNeeded(credentialId, ownerUserId, requestId)) ?? undefined
+}
+
+/**
+ * Whether an inbound payload qualifies for the synchronous ingest-side loading
+ * modal: the trigger opted in, the payload is an interactive type that carries
+ * a fresh usable trigger_id, and the interaction is not already inside a modal.
+ */
+function shouldOpenLoadingModal(
+  body: unknown,
+  providerConfig: Record<string, unknown>
+): body is Record<string, unknown> {
+  if (providerConfig.openLoadingModal !== true) return false
+  if (!isRecordLike(body)) return false
+  if (body.event !== undefined) return false
+  if (typeof body.type !== 'string' || !SLACK_LOADING_MODAL_TYPES.has(body.type)) return false
+  if (typeof body.trigger_id !== 'string' || body.trigger_id === '') return false
+  if (isRecordLike(body.view)) return false
+  return true
+}
+
+/** Fixed Block Kit loading view; only the title and body text are configurable. */
+function buildLoadingView(providerConfig: Record<string, unknown>): Record<string, unknown> {
+  const configuredTitle =
+    typeof providerConfig.loadingModalTitle === 'string' ? providerConfig.loadingModalTitle : ''
+  const configuredText =
+    typeof providerConfig.loadingModalText === 'string' ? providerConfig.loadingModalText : ''
+  const title = configuredTitle.trim() !== '' ? configuredTitle : DEFAULT_LOADING_MODAL_TITLE
+  const text = configuredText.trim() !== '' ? configuredText : DEFAULT_LOADING_MODAL_TEXT
+  return {
+    type: 'modal',
+    callback_id: SLACK_LOADING_MODAL_CALLBACK_ID,
+    /** `truncate` appends its suffix after the slice, so slice to cap − 1. */
+    title: { type: 'plain_text', text: truncate(title, SLACK_MODAL_TITLE_MAX_CHARS - 1, '…') },
+    close: { type: 'plain_text', text: 'Close' },
+    blocks: [
+      {
+        type: 'section',
+        text: {
+          type: 'plain_text',
+          text: truncate(text, SLACK_SECTION_TEXT_MAX_CHARS - 1, '…'),
+          emoji: true,
+        },
+      },
+    ],
+  }
+}
+
+/**
+ * Opens the loading modal via views.open while the trigger_id is fresh. Returns
+ * the created view id, or null on ANY failure — an expired/exchanged trigger_id,
+ * a missing scope, a timeout — because a failed modal must never block dispatch.
+ */
+async function openSlackLoadingModal(
+  botToken: string,
+  triggerId: string,
+  view: Record<string, unknown>,
+  requestId: string
+): Promise<string | null> {
+  try {
+    const response = await fetch('https://slack.com/api/views.open', {
+      method: 'POST',
+      headers: {
+        'Content-Type': 'application/json; charset=utf-8',
+        Authorization: `Bearer ${botToken}`,
+      },
+      body: JSON.stringify({ trigger_id: triggerId, view }),
+      signal: AbortSignal.timeout(SLACK_VIEWS_OPEN_TIMEOUT_MS),
+    })
+    const data = (await response.json()) as { ok?: boolean; error?: string; view?: { id?: string } }
+    if (!data.ok || !data.view?.id) {
+      logger.warn(`[${requestId}] Slack views.open for the loading modal was rejected`, {
+        error: data.error ?? `HTTP ${response.status}`,
+      })
+      return null
+    }
+    return data.view.id
+  } catch (error) {
+    logger.warn(`[${requestId}] Slack views.open for the loading modal failed`, {
+      error: toError(error).message,
+    })
+    return null
+  }
+}
+
 export const slackHandler: WebhookProviderHandler = {
   verifyAuth({ request, rawBody, requestId, providerConfig }: AuthContext) {
     const signingSecret = providerConfig.signingSecret as string | undefined
@@ -863,6 +1021,38 @@ export const slackHandler: WebhookProviderHandler = {
     return new NextResponse(null, { status: 200 })
   },
 
+  /**
+   * Opens the opt-in loading modal while the interaction's 3-second trigger_id
+   * is still fresh — the only window in which a modal can be opened at all,
+   * since the workflow itself runs after the ack. The created view id rides the
+   * execution payload into the trigger output as `loading_view_id` for the
+   * workflow to views.update. Only the credential-based `slack_oauth` trigger
+   * exposes the opt-in, so token resolution here never needs env-var expansion.
+   */
+  async prepareSyncDispatch({
+    body,
+    requestId,
+    providerConfig,
+  }: PrepareSyncDispatchContext): Promise<PrepareSyncDispatchResult | null> {
+    if (!shouldOpenLoadingModal(body, providerConfig)) {
+      return null
+    }
+    const botToken = await resolveSlackWebhookBotToken(providerConfig, requestId)
+    if (!botToken) {
+      logger.warn(
+        `[${requestId}] Loading modal is enabled but no Slack bot token resolved; skipping`
+      )
+      return null
+    }
+    const loadingViewId = await openSlackLoadingModal(
+      botToken,
+      String(body.trigger_id),
+      buildLoadingView(providerConfig),
+      requestId
+    )
+    return loadingViewId ? { syncInteraction: { loadingViewId } } : null
+  },
+
   /**
    * Routes across Slack's three distinct payload families, each identified by
    * a different shape: slash commands (flat form fields with a leading-slash
@@ -870,34 +1060,20 @@ export const slackHandler: WebhookProviderHandler = {
    * `actions[]` and no Events-API `event` envelope), and the Events API
    * (app_mention, message, reaction_added, ... nested under `event`).
    */
-  async formatInput({ body, webhook, requestId }: FormatInputContext): Promise<FormatInputResult> {
+  async formatInput({
+    body,
+    webhook,
+    requestId,
+    credentialOwnerUserId,
+    syncInteraction,
+  }: FormatInputContext): Promise<FormatInputResult> {
     const b = isRecordLike(body) ? body : {}
     const providerConfig = (webhook.providerConfig as Record<string, unknown>) || {}
-    let botToken = providerConfig.botToken as string | undefined
-    // Reusable custom Slack bot credential: use its stored bot token directly.
-    if (!botToken && typeof providerConfig.credentialId === 'string') {
-      const botCredential = await getSlackBotCredential(providerConfig.credentialId)
-      if (botCredential) botToken = botCredential.botToken
-    }
-    // Native (slack_app) triggers carry an OAuth credential rather than a pasted
-    // bot token; resolve it via the credential's OWNER (not the execution actor
-    // in workflow.userId, who may not own the credential) so reaction-message
-    // text and file downloads work.
-    if (!botToken && typeof providerConfig.credentialId === 'string') {
-      const credentialId = providerConfig.credentialId
-      const resolved = await resolveOAuthAccountId(credentialId)
-      if (resolved?.accountId) {
-        const [owner] = await db
-          .select({ userId: account.userId })
-          .from(account)
-          .where(eq(account.id, resolved.accountId))
-          .limit(1)
-        if (owner?.userId) {
-          botToken =
-            (await refreshAccessTokenIfNeeded(credentialId, owner.userId, requestId)) ?? undefined
-        }
-      }
-    }
+    const botToken = await resolveSlackWebhookBotToken(
+      providerConfig,
+      requestId,
+      credentialOwnerUserId
+    )
     const includeFiles = Boolean(providerConfig.includeFiles)
 
     if (typeof b?.command === 'string' && b.command.startsWith('/')) {
@@ -919,7 +1095,11 @@ export const slackHandler: WebhookProviderHandler = {
       ((typeof b?.type === 'string' && SLACK_INTERACTIVE_TYPES.has(b.type)) ||
         Array.isArray(b?.actions))
     ) {
-      return { input: { event: formatSlackInteractive(b) } }
+      const event = formatSlackInteractive(b)
+      if (syncInteraction?.loadingViewId) {
+        event.loading_view_id = syncInteraction.loadingViewId
+      }
+      return { input: { event } }
     }
 
     const rawEvent = b?.event as Record<string, unknown> | undefined
diff --git a/apps/sim/lib/webhooks/providers/types.ts b/apps/sim/lib/webhooks/providers/types.ts
index 64da7506f3c..bf734436af3 100644
--- a/apps/sim/lib/webhooks/providers/types.ts
+++ b/apps/sim/lib/webhooks/providers/types.ts
@@ -38,6 +38,14 @@ export interface FormatInputContext {
   /** HTTP method of the delivering request. Empty on legacy queued jobs. */
   method: string
   requestId: string
+  /**
+   * Owner of the webhook's OAuth credential, when the worker already resolved
+   * it for execution metadata. Lets a provider skip re-resolving the same
+   * credential → account → owner chain; absent on paths that never resolved it.
+   */
+  credentialOwnerUserId?: string
+  /** Interaction context created synchronously at ingest (see {@link SyncInteractionContext}). */
+  syncInteraction?: SyncInteractionContext
 }
 
 /** Result of custom input preparation. */
@@ -46,6 +54,31 @@ export interface FormatInputResult {
   skip?: { message: string }
 }
 
+/**
+ * Interaction context created synchronously at ingest before enqueue, e.g. a
+ * Slack loading modal opened while the interaction's trigger_id was still
+ * fresh. Persisted with the job payload, so it carries identifiers only —
+ * never token material.
+ */
+export interface SyncInteractionContext {
+  /** Slack view id of the loading modal opened via views.open on the ingest path. */
+  loadingViewId: string
+}
+
+/** Context for synchronous pre-enqueue dispatch preparation. */
+export interface PrepareSyncDispatchContext {
+  webhook: Record<string, unknown>
+  workflow: { id: string; userId: string; workspaceId?: string | null }
+  body: unknown
+  requestId: string
+  providerConfig: Record<string, unknown>
+}
+
+/** Result of synchronous pre-enqueue dispatch preparation. */
+export interface PrepareSyncDispatchResult {
+  syncInteraction?: SyncInteractionContext
+}
+
 /** Context for provider-specific file processing before execution. */
 export interface ProcessFilesContext {
   input: Record<string, unknown>
@@ -165,6 +198,15 @@ export interface WebhookProviderHandler {
   /** Custom error response when queuing fails. Return null for default 500. */
   formatQueueErrorResponse?(): NextResponse | null
 
+  /**
+   * Provider work that must run synchronously on the ingest path after admission
+   * and immediately before the execution is enqueued — e.g. opening a Slack
+   * loading modal while the interaction's 3-second trigger_id is still fresh.
+   * Implementations must swallow provider-side failures and return null; the
+   * processor additionally guards the call so a throw can never fail dispatch.
+   */
+  prepareSyncDispatch?(ctx: PrepareSyncDispatchContext): Promise<PrepareSyncDispatchResult | null>
+
   /** Custom input preparation. When defined, replaces the default pass-through of the raw body. */
   formatInput?(ctx: FormatInputContext): Promise<FormatInputResult>
 
diff --git a/apps/sim/lib/workflows/custom-blocks/operations.ts b/apps/sim/lib/workflows/custom-blocks/operations.ts
index 25fabe37bcc..dbb13f87cdf 100644
--- a/apps/sim/lib/workflows/custom-blocks/operations.ts
+++ b/apps/sim/lib/workflows/custom-blocks/operations.ts
@@ -38,11 +38,11 @@ async function eligibleOrgForWorkspace(
 ): Promise<string | null> {
   const ws = await getWorkspaceWithOwner(workspaceId, { includeArchived: true })
   if (!ws?.organizationId) return null
-  if (!(await isFeatureEnabled('deploy-as-block', { userId, orgId: ws.organizationId }))) {
-    return null
-  }
-  if (!(await isOrganizationOnEnterprisePlan(ws.organizationId))) return null
-  return ws.organizationId
+  const [flagEnabled, onEnterprisePlan] = await Promise.all([
+    isFeatureEnabled('deploy-as-block', { userId, orgId: ws.organizationId }),
+    isOrganizationOnEnterprisePlan(ws.organizationId),
+  ])
+  return flagEnabled && onEnterprisePlan ? ws.organizationId : null
 }
 
 /**
diff --git a/apps/sim/lib/workflows/executor/execution-core.ts b/apps/sim/lib/workflows/executor/execution-core.ts
index 9f2ae49e0de..c9dbbba0241 100644
--- a/apps/sim/lib/workflows/executor/execution-core.ts
+++ b/apps/sim/lib/workflows/executor/execution-core.ts
@@ -525,9 +525,25 @@ async function executeWorkflowCoreImpl(
       }
     }
 
-    const [workflowState, env] = await Promise.all([
+    /**
+     * Resolves the org/workspace PII redaction row once for this run; serves both
+     * the input stage and the block-outputs stage (threaded into the executor).
+     * Depends only on the workspace id, so it loads alongside the state and env.
+     */
+    const loadPiiRedactionRow = async () => {
+      const [row] = await db
+        .select({ orgSettings: organization.dataRetentionSettings })
+        .from(workspace)
+        .leftJoin(organization, eq(organization.id, workspace.organizationId))
+        .where(eq(workspace.id, providedWorkspaceId))
+        .limit(1)
+      return row
+    }
+
+    const [workflowState, env, piiRedactionRow] = await Promise.all([
       loadWorkflowState(),
       getExecutionEnvironment(personalEnvUserId, workspaceEnvUserId, providedWorkspaceId),
+      loadPiiRedactionRow(),
     ])
 
     const { blocks, loops, parallels } = workflowState
@@ -810,22 +826,14 @@ async function executeWorkflowCoreImpl(
       allowLargeValueWorkflowScope,
     })
 
-    // Resolve the org/workspace PII redaction policy once; serves both the input
-    // stage (below) and the block-outputs stage (threaded into the executor).
-    // Resolved from stored rules UNCONDITIONALLY — deliberately NOT gated on the
-    // `pii-redaction` feature flag. The flag gates configuration (the settings
-    // route); a transient/false flag read at execution time would skip masking
-    // and leak PII (fail-open). Stored rules are only writable by entitled orgs,
-    // so their presence is the source of truth; absence yields the disabled
-    // default (one indexed lookup, no masking cost for non-PII orgs).
-    const [row] = await db
-      .select({ orgSettings: organization.dataRetentionSettings })
-      .from(workspace)
-      .leftJoin(organization, eq(organization.id, workspace.organizationId))
-      .where(eq(workspace.id, providedWorkspaceId))
-      .limit(1)
+    // The policy applies from stored rules UNCONDITIONALLY — deliberately NOT
+    // gated on the `pii-redaction` feature flag. The flag gates configuration
+    // (the settings route); a transient/false flag read at execution time would
+    // skip masking and leak PII (fail-open). Stored rules are only writable by
+    // entitled orgs, so their presence is the source of truth; absence yields
+    // the disabled default (one indexed lookup, no masking cost for non-PII orgs).
     const piiRedaction: EffectivePiiRedaction = resolveEffectivePiiRedaction({
-      orgSettings: row?.orgSettings,
+      orgSettings: piiRedactionRow?.orgSettings,
       workspaceId: providedWorkspaceId,
     })
 
diff --git a/apps/sim/lib/workflows/persistence/utils.test.ts b/apps/sim/lib/workflows/persistence/utils.test.ts
index ca54ac57740..bea6883e55d 100644
--- a/apps/sim/lib/workflows/persistence/utils.test.ts
+++ b/apps/sim/lib/workflows/persistence/utils.test.ts
@@ -1389,6 +1389,78 @@ describe('Database Helpers', () => {
       expect(dbChainMockFns.where).toHaveBeenCalledTimes(1)
     })
 
+    it('serves a warm admitted version from the cache without the version SELECT', async () => {
+      queueTableRows(schemaMock.workflowDeploymentVersion, [
+        { id: 'dv-warm', state: buildDeployedState() },
+      ])
+
+      await dbHelpers.loadWorkflowDeploymentVersionState('wf-warm', 'dv-warm', 'workspace-1')
+      const second = await dbHelpers.loadWorkflowDeploymentVersionState(
+        'wf-warm',
+        'dv-warm',
+        'workspace-1'
+      )
+
+      expect(second.deploymentVersionId).toBe('dv-warm')
+      expect(dbChainMockFns.where).toHaveBeenCalledTimes(1)
+      expect(mockSanitizeAgentToolsInBlocks).toHaveBeenCalledTimes(1)
+    })
+
+    it('does not serve a cached version to a different workflow id', async () => {
+      queueTableRows(schemaMock.workflowDeploymentVersion, [
+        { id: 'dv-mine', state: buildDeployedState() },
+      ])
+      await dbHelpers.loadWorkflowDeploymentVersionState('wf-mine', 'dv-mine', 'workspace-1')
+
+      await expect(
+        dbHelpers.loadWorkflowDeploymentVersionState('wf-other', 'dv-mine', 'workspace-1')
+      ).rejects.toThrow('Deployment dv-mine was not found for workflow wf-other')
+    })
+
+    it('blockExistsInDeployment answers from the admitted version and warms the cache', async () => {
+      queueTableRows(schemaMock.workflowDeploymentVersion, [
+        { id: 'dv-block', state: buildDeployedState() },
+      ])
+
+      await expect(
+        dbHelpers.blockExistsInDeployment('wf-block', 'block-1', {
+          deploymentVersionId: 'dv-block',
+          workspaceId: 'workspace-1',
+        })
+      ).resolves.toBe(true)
+
+      const deployed = await dbHelpers.loadWorkflowDeploymentVersionState(
+        'wf-block',
+        'dv-block',
+        'workspace-1'
+      )
+      expect(deployed.blocks['block-1']).toBeDefined()
+      expect(dbChainMockFns.where).toHaveBeenCalledTimes(1)
+
+      await expect(
+        dbHelpers.blockExistsInDeployment('wf-block', 'missing-block', {
+          deploymentVersionId: 'dv-block',
+          workspaceId: 'workspace-1',
+        })
+      ).resolves.toBe(false)
+    })
+
+    it('blockExistsInDeployment answers false when the admitted version is missing', async () => {
+      await expect(
+        dbHelpers.blockExistsInDeployment('wf-x', 'block-1', {
+          deploymentVersionId: 'dv-missing',
+          workspaceId: 'workspace-1',
+        })
+      ).resolves.toBe(false)
+    })
+
+    it('blockExistsInDeployment falls back to the raw active-version read without a version id', async () => {
+      queueTableRows(schemaMock.workflowDeploymentVersion, [{ state: buildDeployedState() }])
+
+      await expect(dbHelpers.blockExistsInDeployment('wf-raw', 'block-1')).resolves.toBe(true)
+      expect(mockSanitizeAgentToolsInBlocks).not.toHaveBeenCalled()
+    })
+
     it('invalidateDeployedStateCache(id) forces a rebuild on the next call', async () => {
       queueActiveVersion('dv-inv', buildDeployedState())
       queueActiveVersion('dv-inv', buildDeployedState())
diff --git a/apps/sim/lib/workflows/persistence/utils.ts b/apps/sim/lib/workflows/persistence/utils.ts
index 4aa02891dff..4bb9d5bcfb6 100644
--- a/apps/sim/lib/workflows/persistence/utils.ts
+++ b/apps/sim/lib/workflows/persistence/utils.ts
@@ -94,11 +94,29 @@ export interface DeployedWorkflowData extends NormalizedWorkflowData {
   variables?: Record<string, unknown>
 }
 
+/**
+ * Answers whether a trigger block exists in the workflow's deployment. When the
+ * caller already knows the admitted `deploymentVersionId` (webhook rows carry
+ * it), the check goes through the LRU-backed version loader — warming the cache
+ * the execution path reads moments later — instead of re-reading the full state
+ * jsonb for one boolean. Without an id it falls back to the raw active-version
+ * read. Any failure answers `false`, matching the historical contract.
+ */
 export async function blockExistsInDeployment(
   workflowId: string,
-  blockId: string
+  blockId: string,
+  options?: { deploymentVersionId?: string | null; workspaceId?: string | null }
 ): Promise<boolean> {
   try {
+    if (options?.deploymentVersionId) {
+      const deployed = await loadWorkflowDeploymentVersionState(
+        workflowId,
+        options.deploymentVersionId,
+        options.workspaceId ?? undefined
+      )
+      return Boolean(deployed.blocks[blockId])
+    }
+
     const [result] = await db
       .select({ state: workflowDeploymentVersion.state })
       .from(workflowDeploymentVersion)
@@ -131,10 +149,12 @@ const DEPLOYED_STATE_CACHE_TTL_MS = 5 * 60 * 1000
  * absolute on purpose — it bounds the one non-immutable part, the live credential
  * remap in `applyBlockMigrations` — so credential changes still propagate.
  */
-const deployedStateCache = new LRUCache<string, DeployedWorkflowData>({
-  max: DEPLOYED_STATE_CACHE_MAX_ENTRIES,
-  ttl: DEPLOYED_STATE_CACHE_TTL_MS,
-})
+const deployedStateCache = new LRUCache<string, { workflowId: string; data: DeployedWorkflowData }>(
+  {
+    max: DEPLOYED_STATE_CACHE_MAX_ENTRIES,
+    ttl: DEPLOYED_STATE_CACHE_TTL_MS,
+  }
+)
 
 /** Evicts one deployed-state entry, or clears the cache when no id is given. */
 export function invalidateDeployedStateCache(deploymentVersionId?: string): void {
@@ -191,8 +211,8 @@ export async function materializeDeploymentState(
   executor?: DbOrTx
 ): Promise<DeployedWorkflowData> {
   const cached = deployedStateCache.get(version.id)
-  if (cached) {
-    return structuredClone(cached)
+  if (cached?.workflowId === workflowId) {
+    return structuredClone(cached.data)
   }
 
   const state = version.state as WorkflowState & { variables?: Record<string, unknown> }
@@ -241,7 +261,7 @@ export async function materializeDeploymentState(
     deploymentVersionId: version.id,
   }
 
-  deployedStateCache.set(version.id, deployedState)
+  deployedStateCache.set(version.id, { workflowId, data: deployedState })
   return structuredClone(deployedState)
 }
 
@@ -283,12 +303,21 @@ export async function loadDeployedWorkflowState(
 
 /**
  * Loads an immutable deployment snapshot by ID for work admitted before a later cutover.
+ *
+ * Cache-first: the id is immutable, so a warm LRU entry (guarded by matching
+ * `workflowId`) is byte-identical to what the SELECT + materialization below
+ * would produce, minus the full-state jsonb round trip.
  */
 export async function loadWorkflowDeploymentVersionState(
   workflowId: string,
   deploymentVersionId: string,
   providedWorkspaceId?: string
 ): Promise<DeployedWorkflowData> {
+  const cached = deployedStateCache.get(deploymentVersionId)
+  if (cached?.workflowId === workflowId) {
+    return structuredClone(cached.data)
+  }
+
   const [version] = await db
     .select({
       id: workflowDeploymentVersion.id,
diff --git a/apps/sim/triggers/slack/oauth.ts b/apps/sim/triggers/slack/oauth.ts
index fb277ef8594..f80da71824d 100644
--- a/apps/sim/triggers/slack/oauth.ts
+++ b/apps/sim/triggers/slack/oauth.ts
@@ -20,6 +20,8 @@ const INTERACTION_FILTER_EVENTS = slackEventsSupportingFilter('interaction')
 // Bot/own toggles gate UI visibility only (the route applies them unconditionally),
 // so they are not catalog `filters`.
 const BOT_FILTER_EVENTS = ['message', 'app_mention']
+// Loading-modal opt-in gates UI visibility only; the ingest hook re-checks payload shape.
+const LOADING_MODAL_EVENTS = ['block_actions']
 const OWN_MESSAGE_EVENTS = ['message', 'app_mention', 'reaction_added', 'reaction_removed']
 
 /**
@@ -171,6 +173,45 @@ export const slackOAuthTrigger: TriggerConfig = {
       mode: 'trigger',
       condition: { field: 'eventType', value: INTERACTION_FILTER_EVENTS },
     },
+    {
+      id: 'openLoadingModal',
+      title: 'Open loading modal',
+      type: 'switch',
+      defaultValue: false,
+      description:
+        'Immediately open a loading modal when the interaction arrives, before the workflow runs. Slack trigger IDs expire 3 seconds after the click, so opening a modal from inside the workflow is unreliable — update this one via its loading_view_id output instead.',
+      required: false,
+      mode: 'trigger',
+      condition: { field: 'eventType', value: LOADING_MODAL_EVENTS },
+    },
+    {
+      id: 'loadingModalTitle',
+      title: 'Loading modal title',
+      type: 'short-input',
+      placeholder: 'Working on it',
+      description: 'Title of the loading modal (max 24 characters).',
+      required: false,
+      mode: 'trigger',
+      condition: {
+        field: 'eventType',
+        value: LOADING_MODAL_EVENTS,
+        and: { field: 'openLoadingModal', value: true },
+      },
+    },
+    {
+      id: 'loadingModalText',
+      title: 'Loading modal text',
+      type: 'short-input',
+      placeholder: 'This will just take a moment…',
+      description: 'Body text shown in the loading modal.',
+      required: false,
+      mode: 'trigger',
+      condition: {
+        field: 'eventType',
+        value: LOADING_MODAL_EVENTS,
+        and: { field: 'openLoadingModal', value: true },
+      },
+    },
     {
       id: 'filterBotMessages',
       title: 'Ignore bot messages',
diff --git a/apps/sim/triggers/slack/shared.ts b/apps/sim/triggers/slack/shared.ts
index eda88da4c08..79a20886812 100644
--- a/apps/sim/triggers/slack/shared.ts
+++ b/apps/sim/triggers/slack/shared.ts
@@ -108,6 +108,11 @@ export const SLACK_TRIGGER_OUTPUTS: Record<string, TriggerOutput> = {
         description:
           'Short-lived trigger ID used to open a modal in response. Present for interactivity and slash commands',
       },
+      loading_view_id: {
+        type: 'string',
+        description:
+          'View ID of the loading modal opened at ingest when "Open loading modal" is enabled. Pass it to Slack Update View to replace the loading content — view IDs never expire, unlike the 3-second trigger_id. Empty when the option is off or the open failed',
+      },
       callback_id: {
         type: 'string',
         description: